linse/sics
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- simplified devser queue

Browse Source 
- added inherited sct variables starting with @
- inserted SctIsPending / DevIsPending
- some general improvements in scriptcontext.c/sctdriveobj.c
This commit is contained in:
zolliker
2010-01-27 13:39:33 +00:00
parent 7976b52212
commit a56838b90d
5 changed files with 303 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

394
devser.c
View File

@ -8,24 +8,16 @@ typedef struct DevAction {
DevActionHandler *hdl;
DevPrio prio;
DevKillActionData *kill;
double interval; /* -1 for a queued action */
double timeDue; /* 0 for a queued action */
} DevAction;
typedef struct SchedHeader {
struct SchedHeader *next;
DevAction *actions; /* list of actions for given interval and prio */
DevAction *followingAction;
double interval;
double timeDue;
DevPrio prio;
} SchedHeader;
struct DevSer {
Ascon *asyncConn; /* connection */
Ascon *ascon; /* connection */
DevAction *current;
int killCurrent;
DevAction *actions; /* the action queue */
DevAction *toKill; /* list of actions to be killed */
SchedHeader *headers;
DevAction *actions; /* the action queue */
DevAction *toKill; /* list of actions to be killed */
int steps;
};
@ -65,52 +57,81 @@ static void DevFreeActionList(DevAction * actions)
}
}
DevAction *DevNextAction(DevSer * devser)
{
DevPrio prio;
double now;
SchedHeader *header;
devser->current = NULL;
if (devser->actions) {
prio = devser->actions->prio;
static double nextTime(double last,
double lastInterval,
double now,
double interval) {
double base;
/* nextTime gives the next possible time according to the
* following rules:
* (1) result > now
* (2) result is a multiple of interval
* (3) result > last + min(lastInterval,interval) * 0.99
*/
if (interval > lastInterval) {
base = last + lastInterval * 0.99;
} else {
prio = NullPRIO;
base = last + interval * 0.99;
}
now = DoubleTime();
for (header = devser->headers;
header != NULL && header->prio > prio; header = header->next) {
if (header->followingAction == NULL) {
if (now >= header->timeDue) {
header->followingAction = header->actions;
if (header->interval <= 0) {
header->timeDue = now;
} else {
header->timeDue = (floor(now / header->interval) + 1)
* header->interval;
}
} else if (header->prio == StartPRIO && header->actions != NULL) {
/* special case: poll with StartPRIO pending: block all other actions */
return devser->current;
}
}
if (header->followingAction != NULL) {
devser->current = header->followingAction;
devser->killCurrent = 0;
header->followingAction = header->followingAction->next;
return devser->current;
}
if (now > base) {
base = now;
}
if (devser->actions) {
devser->current = devser->actions;
devser->killCurrent = 1;
devser->actions = devser->actions->next;
}
return devser->current;
return (floor(base / interval) + 1) * interval;
}
int DevQueueTask(void *ds)
static DevAction *DevNextAction(DevSer * devser)
{
/* the action queue is primarily ordered by priority (high prioirty first),
* within one priority by usage (last used action at end)
*/
DevPrio prio;
double now, next;
DevAction *action, **ptr2prev;
devser->current = NULL;
now = DoubleTime();
for (ptr2prev = &devser->actions, action = devser->actions;
action != NULL;
ptr2prev = &action->next, action = action->next) {
if (now >= action->timeDue) {
/* remove action from queue */
*ptr2prev = action->next;
devser->current = action;
if (action->interval >= 0) {
/* this is a scheduled action, to be preserved after use */
devser->killCurrent = 0;
if (action->interval == 0) {
action->timeDue = now;
} else {
/* increase timeDue according to interval */
action->timeDue = nextTime(action->timeDue, action->interval,
now, action->interval);
}
prio = action->prio;
/* insert devser->current before the next lower priority */
for (action = action->next; /* start after this */
action != NULL && action->prio >= prio;
ptr2prev = &action->next, action = action->next);
/* action is now NULL or the next action with lower priority */
*ptr2prev = devser->current;
devser->current->next = action;
} else {
/* this is a queued action, to be killed after use */
devser->killCurrent = 1;
}
return devser->current;
}
if (action->prio == StartPRIO) {
/* if a poll with StartPRIO is scheduled, block all other actions */
return NULL;
}
}
return NULL;
}
static int DevQueueTask(void *ds)
{
DevSer *devser = ds;
AsconStatus status;
@ -131,11 +152,11 @@ int DevQueueTask(void *ds)
}
while (action != NULL) {
status = AsconTask(devser->asyncConn);
status = AsconTask(devser->ascon);
if (status == AsconFailure) {
replyData = AsconGetError(devser->asyncConn);
replyData = AsconGetError(devser->ascon);
} else if (status == AsconReady) {
replyData = AsconRead(devser->asyncConn);
replyData = AsconRead(devser->ascon);
} else {
return 1;
}
@ -144,7 +165,7 @@ int DevQueueTask(void *ds)
}
sendData = action->hdl(action->data, replyData, (status == AsconFailure));
if (sendData != NULL) {
AsconWrite(devser->asyncConn, sendData, 0);
AsconWrite(devser->ascon, sendData, 0);
return 1;
}
if (devser->killCurrent) {
@ -162,19 +183,18 @@ int DevQueueTask(void *ds)
DevSer *DevMake(SConnection * con, int argc, char *argv[])
{
DevSer *devser = NULL;
Ascon *asyncConn = NULL;
Ascon *ascon = NULL;
asyncConn = AsconMake(con, argc, argv);
if (!asyncConn) {
ascon = AsconMake(con, argc, argv);
if (!ascon) {
return NULL;
}
devser = calloc(1, sizeof(*devser));
assert(devser);
devser->asyncConn = asyncConn;
devser->ascon = ascon;
devser->current = NULL;
devser->killCurrent = 0;
devser->actions = NULL;
devser->headers = NULL;
devser->toKill = NULL;
devser->steps = -1; /* no debugging by default */
TaskRegister(pServ->pTasker, DevQueueTask, NULL, NULL, devser, 0);
@ -186,38 +206,17 @@ void DevDebugMode(DevSer * devser, int steps)
devser->steps = steps;
}
DevAction *DevNewAction(void *data, DevActionHandler hdl,
DevKillActionData * killFunc, DevPrio prio)
{
DevAction *action;
action = calloc(1, sizeof(*action));
assert(action);
action->data = data;
action->hdl = hdl;
action->kill = killFunc;
action->prio = prio;
action->next = NULL;
return action;
}
void DevKill(DevSer * devser)
{
SchedHeader *h, *victim;
if (devser->asyncConn) {
AsconKill(devser->asyncConn);
if (devser->ascon) {
AsconKill(devser->ascon);
}
DevFreeActionList(devser->actions);
DevFreeActionList(devser->toKill);
h = devser->headers;
while (h != NULL) {
victim = h;
h = victim->next;
DevFreeActionList(victim->actions);
free(victim);
}
if (devser->killCurrent) {
if (devser->current->kill != NULL) devser->current->kill(devser->current);
if (devser->current->kill != NULL) {
devser->current->kill(devser->current);
}
devser->killCurrent = 0;
free(devser->current);
}
@ -227,67 +226,35 @@ void DevKill(DevSer * devser)
void DevDisconnect(DevSer * devser)
{
if (devser->asyncConn) {
AsconDisconnect(devser->asyncConn);
if (devser->ascon) {
AsconDisconnect(devser->ascon);
}
}
int DevQueue(DevSer * devser, void *actionData, DevPrio prio,
DevActionHandler hdl, DevActionMatch * matchFunc,
DevKillActionData * killFunc)
int DevUnschedule(DevSer * devser, void *callData,
DevActionHandler * hdl, DevActionMatch * matchFunc)
{
DevAction *action, **ptr2Last;
DevAction *new;
if (prio <= NullPRIO)
prio = NullPRIO + 1;
if (prio >= NumberOfPRIO)
prio = NumberOfPRIO - 1;
ptr2Last = &devser->actions;
for (action = devser->actions; action != NULL && action->prio >= prio;
action = action->next) {
if (action->hdl == hdl && matchFunc(actionData, action->data)) {
return 0; /* there is already an identical action */
}
ptr2Last = &action->next;
}
new = DevNewAction(actionData, hdl, killFunc, prio);
new->next = action;
*ptr2Last = new;
return 1;
}
int DevUnschedule(DevSer * devser, void *actionData,
DevActionHandler hdl, DevActionMatch * matchFunc)
{
SchedHeader *header = NULL;
DevAction **ptr2Last = NULL;
DevAction **ptr2prev = NULL;
DevAction *action = NULL;
int cnt = 0;
/* scan through all headers */
for (header = devser->headers; header != NULL; header = header->next) {
ptr2Last = &header->actions;
for (action = header->actions; action != NULL; action = *ptr2Last) {
if (action->hdl == hdl && matchFunc(actionData, action->data)) {
if (action == header->followingAction) {
/* advance followingAction if equal */
header->followingAction = action->next;
}
if (action == devser->current) {
devser->current = NULL;
devser->killCurrent = 0; /* should already be 0 */
}
cnt++;
/* remove from list */
*ptr2Last = action->next;
/* add to kill list */
action->next = devser->toKill;
devser->toKill = action;
} else {
ptr2Last = &action->next;
/* scan through the queue */
for (ptr2prev = &devser->actions, action = devser->actions;
action != NULL;
action = action->next) {
if (action->hdl == hdl && matchFunc(callData, action->data)) {
if (action == devser->current) {
devser->current = NULL;
devser->killCurrent = 0;
}
cnt++;
/* remove from list */
*ptr2prev = action->next;
/* add to kill list */
action->next = devser->toKill;
devser->toKill = action;
} else {
ptr2prev = &action->next;
}
}
return cnt;
@ -295,67 +262,92 @@ int DevUnschedule(DevSer * devser, void *actionData,
int DevSchedule(DevSer * devser, void *actionData,
DevPrio prio, double interval,
DevActionHandler hdl, DevActionMatch * matchFunc,
DevActionHandler * hdl, DevActionMatch * matchFunc,
DevKillActionData * killFunc)
{
SchedHeader *header = NULL;
SchedHeader **ptr2LastHeader = NULL;
SchedHeader *newHeader;
DevAction *action = NULL;
DevAction **ptr2Last = NULL;
DevAction *newAction;
DevAction *action;
DevAction *foundAction = NULL;
DevAction **ptr2prev;
DevAction **ptr2insertPos = NULL;
int ret;
if (prio <= NullPRIO)
if (prio <= NullPRIO) {
prio = NullPRIO + 1;
if (prio >= NumberOfPRIO)
}
if (prio >= NumberOfPRIO) {
prio = NumberOfPRIO - 1;
ret = DevUnschedule(devser, actionData, hdl, matchFunc);
newAction = DevNewAction(actionData, hdl, killFunc, prio);
/* find matching header */
ptr2LastHeader = &devser->headers;
for (header = devser->headers; header != NULL; header = *ptr2LastHeader) {
if (header->prio == newAction->prio && header->interval == interval) {
/* append new action at the tail */
ptr2Last = &header->actions;
for (action = header->actions; action != NULL; action = action->next) {
ptr2Last = &action->next;
}
*ptr2Last = newAction;
assert(newAction->next == NULL);
return ret;
} else if (header->prio < newAction->prio ||
(header->prio == newAction->prio
&& header->interval > interval)) {
break;
}
for (ptr2prev = &devser->actions, action = devser->actions;
action != NULL;
ptr2prev = &action->next, action = action->next) {
if (action->prio < prio && ptr2insertPos == NULL) {
ptr2insertPos = ptr2prev;
}
if (header->actions == NULL) {
/* remove empty header */
*ptr2LastHeader = header->next;
free(header);
} else {
ptr2LastHeader = &header->next;
/* check if it is the same action (only once) */
if (action->hdl == hdl
&& action->kill == killFunc
&& (interval < 0) == (action->interval < 0)
&& foundAction == NULL
&& matchFunc(actionData, action->data)) {
if (prio == action->prio && interval < 0) {
/* do not move an action with equal prio */
killFunc(actionData);
return 0;
}
/* remove action from list */
*ptr2prev = action->next;
foundAction = action;
}
}
/* insert new header */
newHeader = calloc(1, sizeof(*newHeader));
assert(newHeader);
newHeader->actions = newAction;
newHeader->followingAction = NULL;
newHeader->prio = newAction->prio;
newHeader->interval = interval;
newHeader->next = header;
newHeader->timeDue = DoubleTime() + interval;
*ptr2LastHeader = newHeader;
/* create if needed */
if (foundAction != NULL) {
action = foundAction;
ret = 0;
killFunc(actionData);
} else {
action = calloc(1, sizeof(*action));
assert(action);
action->data = actionData;
action->hdl = hdl;
action->kill = killFunc;
action->timeDue = 0;
ret = 1;
}
action->prio = prio;
/* insert into queue */
if (ptr2insertPos == NULL) {
ptr2insertPos = ptr2prev;
}
action->next = *ptr2insertPos;
*ptr2insertPos = action;
if (interval < 0) { /* case "queued" */
action->interval = -1.0;
} else { /* case "scheduled" */
if (action->timeDue == 0) { /* not yet scheduled: do it immediately */
action->timeDue = DoubleTime();
} else { /* calculate next time */
action->timeDue = nextTime(action->timeDue, action->interval,
DoubleTime(), interval);
}
action->interval = interval;
}
return ret;
}
int DevQueue(DevSer * devser, void *actionData, DevPrio prio,
DevActionHandler * hdl, DevActionMatch * matchFunc,
DevKillActionData * killFunc)
{
return DevSchedule(devser, actionData, prio, -1.0, hdl, matchFunc, killFunc);
}
int DevRemoveAction(DevSer * devser, void *actionData)
{
SchedHeader *header = NULL;
DevAction **ptr2Last = NULL;
DevAction **ptr2prev = NULL;
DevAction *action = NULL;
int cnt = 0;
@ -374,18 +366,32 @@ int DevRemoveAction(DevSer * devser, void *actionData)
devser->current = NULL;
}
/* remove from queue */
ptr2Last = &devser->actions;
for (action = devser->actions; action != NULL; action = *ptr2Last) {
ptr2prev = &devser->actions;
for (action = devser->actions; action != NULL; action = *ptr2prev) {
if (actionData == action->data) {
cnt++;
/* remove from list */
*ptr2Last = action->next;
*ptr2prev = action->next;
if (action->kill != NULL)
action->kill(action->data);
free(action);
} else {
ptr2Last = &action->next;
ptr2prev = &action->next;
}
}
return cnt++;
}
int DevIsPending(DevSer * devser, void *callData,
DevActionHandler * hdl, DevActionMatch * matchFunc)
{
DevAction *action = devser->current;
if (action) {
if (action->hdl == hdl && matchFunc(callData, action->data)) {
return 1;
}
}
return 0;
}