You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
372 lines
16 KiB
372 lines
16 KiB
#include <windows.h>
|
|
#include <string.h>
|
|
#include <stdio.h>
|
|
#include "gutils.h"
|
|
#include "list.h"
|
|
#include "queue.h"
|
|
|
|
#define NAMELENGTH 20
|
|
|
|
typedef struct queue_tag{
|
|
CRITICAL_SECTION CritSect; /* to single-thread queue operations */
|
|
HANDLE Event; /* Event to set when finished */
|
|
HANDLE Sem; /* semaphore for Get to wait on */
|
|
HANDLE MaxSem; /* semaphore for Put to wait on */
|
|
int Waiting; /* num tasks waiting ~= -(Sem count) */
|
|
LIST List; /* the queue itself */
|
|
BOOL Alive; /* TRUE => no destroy request yet */
|
|
BOOL Aborted; /* TRUE => the queue has been aborted */
|
|
EMPTYPROC Emptier; /* the thread proc for emptying */
|
|
int MaxEmptiers; /* max number of emptying threads */
|
|
int MinQueueToStart;/* start another emptier Q this long */
|
|
int MaxQueue; /* absolute maximum size of queue
|
|
* (for debug only) */
|
|
int Running; /* number of emptiers in existence
|
|
* Once an emptier is created this
|
|
* stays positive until Queue_Destroy */
|
|
DWORD InstanceData; /* instance data for emptier */
|
|
char Name[NAMELENGTH+1]; /* Name for the queue (for debug) */
|
|
} QUEUEDATA;
|
|
|
|
/* DOGMA:
|
|
Any Wait must occur outside the critical section.
|
|
|
|
Any update to the queue must occur inside the critical section.
|
|
Any peeking from outside the critical section must be taken with salt.
|
|
|
|
The queue has between 0 and MaxQueue elements on its list. The Maximum
|
|
is policed by MaxSem which is initialised to MaxQueue and Waited for by
|
|
Put before adding an element and Released by Get whenever it takes an element
|
|
off. MaxQueue itself is just kept around for debug purposes.
|
|
|
|
Put must Wait before entering the critical section, therefore a failed Put
|
|
(e.g. Put to an Aborted queue) will have already upset the semaphore and so
|
|
must back it out.
|
|
|
|
Abort clears the queue and so must adequately simulate the elements being
|
|
gotten. In fact it just does a single Release on MaxSem which ensures that
|
|
a Put can complete. Any blocked Puts will then succeed one at a time as
|
|
each one backs out.
|
|
|
|
Abort is primarily intended for use by the Getter. Caling it before any
|
|
element has ever been put is peculiar, but harmless.
|
|
|
|
The minumum is policed by Sem which is initialised to 0, is Waited for by
|
|
Get before getting an element and Released by Put whenever it puts one.
|
|
Queue_Destroy neds to ensure that no thread will block on the Get but all
|
|
threads will run into the empty queue and get STOPTHREAD or ENDQUEUE. It
|
|
therefore releases the semaphore as many times as there are threads running.
|
|
|
|
Abort clears the queue and simulates the elements being gotten so that
|
|
a single Get is left blocked waiting for the Destroy. Whether there is a
|
|
Get actually waiting at the moment is not interesting. Even if there were
|
|
not, one could be by the time the abort is done. There are the following
|
|
cases (Not Alive means the Queue_Destroy is already in):
|
|
Not empty Alive -> empty it, let all but 1 run.
|
|
Not empty Not Alive -> empty it, let all run.
|
|
Empty Alive -> let all but 1 run.
|
|
Empty Not Alive -> let all run.
|
|
Since Queue_Destroy has already released everything, the Not Alive cases
|
|
need no further releasing.
|
|
*/
|
|
|
|
|
|
/* Queue_Create:
|
|
** Return a queue handle for a newly created empty queue
|
|
** NULL returned means it failed.
|
|
*/
|
|
QUEUE Queue_Create( EMPTYPROC Emptier /* thread proc to start */
|
|
, int MaxEmptiers /* max Getting threads */
|
|
, int MinQueueToStart /* elements per thread */
|
|
, int MaxQueue /* max elements on q */
|
|
, HANDLE Event /* signal on deallocation */
|
|
, DWORD InstanceData
|
|
, PSZ Name
|
|
)
|
|
{ QUEUE Queue;
|
|
|
|
Queue = (QUEUE)GlobalAlloc(GMEM_FIXED, sizeof(QUEUEDATA));
|
|
if (Queue==NULL) {
|
|
char msg[80];
|
|
wsprintf(msg, "Could not allocate storage for queue %s", Name);
|
|
/* Trace_Error(msg, FALSE); */
|
|
return NULL;
|
|
}
|
|
InitializeCriticalSection(&Queue->CritSect);
|
|
//??? should allow for failure!!!
|
|
/* the value of about 10 million is chosen to be effectively infinite */
|
|
Queue->Sem = CreateSemaphore(NULL, 0, 99999999, NULL);
|
|
//??? should allow for failure!!!
|
|
Queue->MaxSem = CreateSemaphore(NULL, MaxQueue, 99999999, NULL);
|
|
//??? should allow for failure!!!
|
|
Queue->Waiting = 0;
|
|
Queue->List = List_Create();
|
|
Queue->Alive = TRUE;
|
|
Queue->Aborted = FALSE;
|
|
Queue->Emptier = Emptier;
|
|
Queue->MaxEmptiers = MaxEmptiers;
|
|
Queue->MinQueueToStart = MinQueueToStart;
|
|
Queue->MaxQueue = MaxQueue;
|
|
Queue->Running = 0;
|
|
Queue->Event = Event;
|
|
Queue->InstanceData = InstanceData;
|
|
strncpy(Queue->Name, Name, NAMELENGTH);
|
|
Queue->Name[NAMELENGTH]='\0'; /* guardian */
|
|
return Queue;
|
|
} /* Queue_Create */
|
|
|
|
|
|
/* Destroy:
|
|
** Internal procedure.
|
|
** Actually deallocate the queue and signal its event (if any)
|
|
** Must have already left the critical section
|
|
*/
|
|
static void Destroy(QUEUE Queue)
|
|
{
|
|
//dprintf1(("Actual Destroy of queue '%s'\n", Queue->Name));
|
|
DeleteCriticalSection(&(Queue->CritSect));
|
|
CloseHandle(Queue->Sem);
|
|
CloseHandle(Queue->MaxSem);
|
|
List_Destroy(&(Queue->List));
|
|
if (Queue->Event!=NULL) {
|
|
SetEvent(Queue->Event);
|
|
}
|
|
GlobalFree( (HGLOBAL)Queue);
|
|
} /* Destroy */
|
|
|
|
|
|
/* Queue_Put:
|
|
** Put an element from buffer Data of length Len bytes onto the queue.
|
|
** Will wait until the queue has room
|
|
** FALSE returned means the queue has been aborted and no
|
|
** put will ever succeed again.
|
|
** This operation may NOT be performed after a Queue_Destroy on Queue
|
|
*/
|
|
BOOL Queue_Put(QUEUE Queue, LPBYTE Data, UINT Len)
|
|
{
|
|
DWORD ThreadId;
|
|
//dprintf1(("Put to queue '%s'\n", Queue->Name));
|
|
WaitForSingleObject(Queue->MaxSem, INFINITE);
|
|
EnterCriticalSection(&Queue->CritSect);
|
|
//dprintf1(("Put running to queue '%s'\n", Queue->Name));
|
|
if ((Queue->Aborted) || (!Queue->Alive)) {
|
|
//dprintf1(("(legal) Queue_Put to Aborted queue '%s'\n", Queue->Name));
|
|
LeaveCriticalSection(&Queue->CritSect);
|
|
ReleaseSemaphore(Queue->MaxSem, 1, NULL); /* let next in */
|
|
return FALSE; /* Caller should soon please Queue_Destroy */
|
|
}
|
|
List_AddFirst(Queue->List, Data, Len);
|
|
ReleaseSemaphore(Queue->Sem, 1, NULL);
|
|
--Queue->Waiting;
|
|
if ( Queue->Running < Queue->MaxEmptiers
|
|
&& ( Queue->Running<=0
|
|
|| List_Card(Queue->List) > Queue->MinQueueToStart*Queue->Running
|
|
)
|
|
) {
|
|
++Queue->Running;
|
|
LeaveCriticalSection(&Queue->CritSect);
|
|
return ( (BOOL)CreateThread( NULL
|
|
, 0
|
|
, (LPTHREAD_START_ROUTINE)
|
|
Queue->Emptier
|
|
, (LPVOID)Queue
|
|
, 0
|
|
, &ThreadId
|
|
)
|
|
);
|
|
}
|
|
LeaveCriticalSection(&Queue->CritSect);
|
|
return TRUE;
|
|
} /* Queue_Put */
|
|
|
|
/* Queue_Get:
|
|
** Get an element from the queue. (Waits until there is one)
|
|
** The elemeent is copied into Data. MaxLen is buffer length in bytes.
|
|
** Negative return codes imply no element is gotten.
|
|
** A negative return code is STOPTHREAD or ENDQUEUE or an error.
|
|
** On receiving STOPTHREAD or ENDQUEUE the caller should clean up and
|
|
** then ExitThread(0);
|
|
** If the caller is the last active thread getting from this queue, it
|
|
** will get ENDQUEUE rather than STOPTHREAD.
|
|
** Positive return code = length of data gotten in bytes.
|
|
*/
|
|
int Queue_Get(QUEUE Queue, LPBYTE Data, int MaxLen)
|
|
{ LPBYTE ListData;
|
|
int Len;
|
|
//dprintf1(("Get from queue '%s'\n", Queue->Name));
|
|
EnterCriticalSection(&Queue->CritSect);
|
|
//dprintf1(("Get running from queue '%s'\n", Queue->Name));
|
|
if (List_IsEmpty(Queue->List)) {
|
|
if (!Queue->Alive) {
|
|
--(Queue->Running);
|
|
if (Queue->Running<=0 ) {
|
|
if (Queue->Running<0 ) {
|
|
char msg[80];
|
|
wsprintf( msg
|
|
, "Negative threads running on queue %s"
|
|
, Queue->Name
|
|
);
|
|
// Trace_Error(msg, FALSE);
|
|
// return NEGTHREADS; ???
|
|
}
|
|
LeaveCriticalSection(&Queue->CritSect);
|
|
Destroy(Queue);
|
|
return ENDQUEUE;
|
|
}
|
|
LeaveCriticalSection(&Queue->CritSect);
|
|
return STOPTHREAD;
|
|
}
|
|
if (Queue->Waiting>0) {
|
|
/* already another thread waiting, besides us */
|
|
--(Queue->Running);
|
|
LeaveCriticalSection(&(Queue->CritSect));
|
|
return STOPTHREAD;
|
|
}
|
|
}
|
|
|
|
++(Queue->Waiting);
|
|
LeaveCriticalSection(&(Queue->CritSect));
|
|
WaitForSingleObject(Queue->Sem, INFINITE);
|
|
EnterCriticalSection(&(Queue->CritSect));
|
|
|
|
/* If the queue is empty now it must be dead */
|
|
if (List_IsEmpty(Queue->List)) {
|
|
if (Queue->Alive && (!Queue->Aborted)) {
|
|
char msg[80];
|
|
wsprintf( msg
|
|
, "Queue %s empty but not dead during Get!"
|
|
, Queue->Name
|
|
);
|
|
// Trace_Error(msg, FALSE);
|
|
return SICKQUEUE;
|
|
}
|
|
else {
|
|
--(Queue->Running);
|
|
if (Queue->Running==0) {
|
|
LeaveCriticalSection(&(Queue->CritSect));
|
|
Destroy(Queue);
|
|
return ENDQUEUE;
|
|
}
|
|
LeaveCriticalSection(&(Queue->CritSect));
|
|
return STOPTHREAD;
|
|
}
|
|
}
|
|
|
|
/* The queue is not empty and we are in the critical section. */
|
|
ListData = List_Last(Queue->List);
|
|
Len = List_ItemLength(ListData);
|
|
if (Len>MaxLen) {
|
|
ReleaseSemaphore(Queue->Sem, 1, NULL);
|
|
--Queue->Waiting;
|
|
LeaveCriticalSection(&Queue->CritSect);
|
|
return TOOLONG;
|
|
}
|
|
memcpy(Data, ListData, Len);
|
|
List_DeleteLast(Queue->List);
|
|
LeaveCriticalSection(&Queue->CritSect);
|
|
ReleaseSemaphore(Queue->MaxSem, 1, NULL);
|
|
return Len;
|
|
} /* Queue_Get */
|
|
|
|
|
|
/* Queue_Destroy:
|
|
** Mark the queue as completed. No further data may ever by Put on it.
|
|
** When the last element has been gotten, it will return ENDTHREAD to
|
|
** a Queue_Get and deallocate itself. If it has an Event it will signal
|
|
** the event at that point.
|
|
** The Queue_Destroy operation returns promptly. It does not wait for
|
|
** further Gets or for the deallocation.
|
|
*/
|
|
void Queue_Destroy(QUEUE Queue)
|
|
{
|
|
EnterCriticalSection(&(Queue->CritSect));
|
|
//dprintf1(("Queue_Destroy %s\n", Queue->Name));
|
|
Queue->Alive = FALSE;
|
|
if ( List_IsEmpty(Queue->List)) {
|
|
if (Queue->Running==0) {
|
|
/* Only possible if nobody ever got started */
|
|
LeaveCriticalSection(&(Queue->CritSect));
|
|
Destroy(Queue);
|
|
return;
|
|
}
|
|
else { int i;
|
|
/* The list is empty, but some threads could be
|
|
blocked on the Get (or about to block) so
|
|
release every thread that might ever wait on Get */
|
|
for (i=0; i<Queue->Running; ++i) {
|
|
ReleaseSemaphore(Queue->Sem, 1, NULL);
|
|
--(Queue->Waiting);
|
|
}
|
|
LeaveCriticalSection(&(Queue->CritSect));
|
|
}
|
|
}
|
|
else LeaveCriticalSection(&(Queue->CritSect));
|
|
return;
|
|
} /* Queue_Destroy */
|
|
|
|
/* Queue_GetInstanceData:
|
|
** Retrieve the DWORD of instance data that was given on Create
|
|
*/
|
|
DWORD Queue_GetInstanceData(QUEUE Queue)
|
|
{ return Queue->InstanceData;
|
|
} /* Queue_GetInstanceData */
|
|
|
|
/* Queue_Abort:
|
|
** Abort the queue. Normally called by the Getter.
|
|
** Discard all elements on the queue,
|
|
** If the queue has already been aborted this will be a no-op.
|
|
** It purges all the data elements. If the Abort parameter is non-NULL
|
|
** then it is called for each element before deallocating it. This
|
|
** allows storage which is hung off the element to be freed.
|
|
** After this, all Put operations will return FALSE. If they were
|
|
** waiting they will promptly complete. The queue is NOT deallocated.
|
|
** That only happens when the last Get completes after the queue has been
|
|
** Queue_Destroyed. This means the normal sequence is:
|
|
** Getter discovers that the queue is now pointless and does Queue_Abort
|
|
** Getter does another Get (which blocks)
|
|
** Putter gets FALSE return code on next (or any outstanding) Put
|
|
** (Putter may want to propagates the error back to his source)
|
|
** Putter does Queue_Destroy
|
|
** The blocked Get is released and the queue is deallocated.
|
|
*/
|
|
|
|
void Queue_Abort(QUEUE Queue, QUEUEABORTPROC Abort)
|
|
{
|
|
/* This is similar to Destroy, but we call the Abort proc and
|
|
free the storage of the elements. Destroy allows them to run down.
|
|
|
|
It is essential that the last Get must block until the sender does a
|
|
Queue_Destroy (if it has not been done already). The Alive flag
|
|
tells whether the Queue_Destroy has been done. All Getters except
|
|
the last should be released.
|
|
*/
|
|
//dprintf1(("Queue_Abort '%s'\n", Queue->Name));
|
|
EnterCriticalSection(&(Queue->CritSect));
|
|
//dprintf1(("Queue_Abort running for queue '%s'\n", Queue->Name));
|
|
for (; ; ) {
|
|
LPSTR Cursor = List_First(Queue->List);
|
|
int Len;
|
|
if (Cursor==NULL) break;
|
|
Len = List_ItemLength(Cursor);
|
|
if (Abort!=NULL) {
|
|
Abort(Cursor, Len);
|
|
}
|
|
List_DeleteFirst(Queue->List);
|
|
}
|
|
/* Queue is now empty. Do not destroy. That's for the Putters */
|
|
Queue->Aborted = TRUE;
|
|
|
|
/* make sure the next Queue_Get blocks unless Queue_Destroy already done */
|
|
//dprintf1(("Queue_Abort '%s' fixing semaphore to block\n", Queue->Name));
|
|
if (Queue->Alive){
|
|
while(Queue->Waiting<0) {
|
|
WaitForSingleObject(Queue->Sem, INFINITE);
|
|
++(Queue->Waiting);
|
|
}
|
|
}
|
|
//dprintf1(("Queue_Abort '%s' semaphore now set to block\n", Queue->Name));
|
|
|
|
LeaveCriticalSection(&(Queue->CritSect));
|
|
ReleaseSemaphore(Queue->MaxSem, 1, NULL);
|
|
return;
|
|
} /* Queue_Abort */
|