diff --git a/src/agents/store/Bongo.rules b/src/agents/store/Bongo.rules index f7c148f..fc72901 100644 --- a/src/agents/store/Bongo.rules +++ b/src/agents/store/Bongo.rules @@ -22,6 +22,7 @@ bongostore_SOURCES = \ src/agents/store/db.c \ src/agents/store/dbpool.c \ src/agents/store/db.h \ + src/agents/store/fairlock.c \ src/agents/store/guid.c \ src/agents/store/help.h \ src/agents/store/lock.c \ diff --git a/src/agents/store/command.c b/src/agents/store/command.c index eb1c4fa..d4f2983 100644 --- a/src/agents/store/command.c +++ b/src/agents/store/command.c @@ -4139,7 +4139,7 @@ StoreCommandPROPSET(StoreClient *client, { DStoreDocInfo info; int ccode; - NLockStruct *lock; + NLockStruct *lock = NULL; BongoJsonNode *alarmjson = NULL; switch (DStoreGetDocInfoGuid(client->handle, guid, &info)) { @@ -4233,7 +4233,7 @@ StoreCommandREAD(StoreClient *client, { int ccode = 0; DStoreDocInfo info; - NLockStruct *lock; + NLockStruct *lock = NULL; switch (DStoreGetDocInfoGuid(client->handle, guid, &info)) { case -1: @@ -4794,8 +4794,7 @@ StoreCommandWATCH(StoreClient *client, DStoreDocInfo *collection, int flags) return ccode; } - ccode = StoreGetExclusiveLock(client, &lock, collection->guid, 3000); - // ccode = StoreGetCollectionLock(client, &lock, collection->guid); + ccode = StoreGetCollectionLock(client, &lock, collection->guid); if (ccode) { return ccode; } diff --git a/src/agents/store/fairlock.c b/src/agents/store/fairlock.c new file mode 100644 index 0000000..382fe13 --- /dev/null +++ b/src/agents/store/fairlock.c @@ -0,0 +1,344 @@ +/** \file + * An implementation of locks which governs order of access via a FIFO queue + */ + +#include "stored.h" +#include "lock.h" + +NFairLockPool *lockpool; + +/** + * Initialize the pool of fair locks. Needs to be called once per program + * in order for fair locks to be available + */ +void +StoreInitializeFairLocks() +{ + lockpool = MemMalloc(sizeof(NFairLockPool)); + XplOpenLocalSemaphore(lockpool->lock, SEM_UNLOCKED); + lockpool->first = NULL; + lockpool->last = NULL; +} + +/** \internal + * Create a string name for a store resource, used for find a specific lock + * for that resource + * \param name Usually the name of a store the resource is in + * \param guid GUID of the resource within that store + * \return A string name unique to that resource + */ +char * +FairLockName(char *name, uint64_t guid) +{ + char *result; + int result_len; + + result_len = strlen(name) + 33; + result = MemMalloc(result_len); + + snprintf(result, result_len, "%s-" GUID_FMT, name, guid); + + return result; +} + +/** \internal + * Retrieve a lock for a resource from a global pool of locks. This doesn't ascertain + * or set the actual status of that lock, it simply finds a reference to it. + * \bug A long list is a bad implementation, we want a big hash of short lists for quick access + * \bug Should we retire locks who haven't been taken in a while while we're looking? + * \param store Usually the name of the store the resource resides. + * \param guid The GUID of the resource within that store + * \return A fair lock governing access to that resource + */ +NFairLock * +FairLockFind(char *store, uint64_t guid) +{ + NFairLockList *search; + NFairLock *result; + char *name; + char *new_lock = ""; + + name = FairLockName(store, guid); + XplWaitOnLocalSemaphore(lockpool->lock); + + // lock for our lock + result = NULL; + search = lockpool->first; + while ((search != NULL) && (result == NULL)) { + if (strcmp(search->lock->lockname, name) == 0) { + result = search->lock; + } + search = search->next; + } + + if (result == NULL) { + // need to add a new lock + NFairLockList *new_entry; + new_lock = "new "; + + new_entry = MemMalloc(sizeof(NFairLockList)); + result = MemMalloc(sizeof(NFairLock)); + + new_entry->next = NULL; + new_entry->lock = result; + + result->lockname = MemMalloc(strlen(name)+1); + result->readers = 0; + result->writer = 0; + result->queue = NULL; + XplOpenLocalSemaphore(result->access, SEM_UNLOCKED); + strcpy(result->lockname, name); + + if (lockpool->first != NULL) { + // append to existing list + lockpool->last->next = new_entry; + lockpool->last = new_entry; + } else { + // start a new list + lockpool->first = new_entry; + lockpool->last = new_entry; + } + } + + XplSignalLocalSemaphore(lockpool->lock); + MemFree(name); + return result; +} + +/** \internal + * Places a ticket in the queue for a lock. Usually called when someone already + * has the lock, and the process wishes to queue up for when it's free again. + * \param lock Fair lock we want access to + * \param is_writer Whether or not we desire write access to the resource + * \return A ticket which has been placed in the queue for the lock + */ +NFairLockQueueEntry * +FairLockNewQueueEntry (NFairLock *lock, BOOL is_writer) +{ + NFairLockQueueEntry *result; + + result = MemMalloc(sizeof(NFairLockQueueEntry)); + if (result == NULL) return NULL; + + result->is_writer = is_writer; + result->next = NULL; + XplOpenLocalSemaphore(result->signal, SEM_LOCKED); + + if (lock->queue == NULL) { + // first entry in the queue + lock->queue = result; + } else { + // find the end of the queue + NFairLockQueueEntry *list = lock->queue; + + while(list->next != NULL) { list = list->next; } + list->next = result; + } + return (result); +} + +/** \internal + * Remove a ticket from the from of the queue for a lock. This is called + * by the process holding the ticket after it has acquired the lock, + * since the presence of the queue on the lock prevents other threads + * stealing the lock. + * The ticket itself isn't provided, we assume it's simply at the front + * of the queue. + * \bug Is the above assumption right? + * \param lock Fair lock whose queue the ticket was in. + */ +void +FairLockRemoveQueueHead (NFairLock *lock) +{ + NFairLockQueueEntry *entry; + + if (lock->queue == NULL) return; // no queue to alter? + entry = lock->queue; + + lock->queue = entry->next; + XplCloseLocalSemaphore(entry->signal); + MemFree(entry); +} + +/** \internal + * Signal the thread who holds the ticket at the front of the queue for a + * fair lock that it's likely it can now proceed to gain the lock. Doesn't + * touch the queue itself, as removing it too early could allow another thread + * to jump the queue. + * \param lock Fair lock whose queue we want to look at. + */ +void +FairLockSignalQueueHead (NFairLock *lock) +{ + if (lock->queue == NULL) return; // no-one to signal + + if (lock->queue->is_writer == FALSE) { + // a thread is waiting for a read lock + XplSignalLocalSemaphore(lock->queue->signal); + XplSignalLocalSemaphore(lock->queue->signal); + } else { + // a thread is waiting for a write lock - only signal if readers have left + // this eventually succeeds when the last read lock is given up + // new acquirees queue until then + if (lock->readers == 0) { + XplSignalLocalSemaphore(lock->queue->signal); + XplSignalLocalSemaphore(lock->queue->signal); + } + } +} + +/** + * Get an Exclusive lock to a store resource. With an exclusive lock, no + * other thread should read from or write to the resource protected. If + * other readers/writer already holds the lock, this will wait in a queue + * until the lock becomes free. + * \bug Timeout is completely useless atm. + * \param client Store client accessing the resource + * \param lock Where we want to store the fair lock we receive + * \param guid GUID of the resource we wish to access in the store + * \param timeout How long we're willing to wait for the lock + * \return 0 if we gained the lock, some error number otherwise. + */ +int +StoreGetExclusiveFairLockQuiet(StoreClient *client, NFairLock **lock, + uint64_t guid, time_t timeout) +{ + NFairLock *found_lock; + int result = 0; + + // find the lock + found_lock = FairLockFind(client->storeName, guid); + if (found_lock == NULL) return -1; // error + *lock = found_lock; + + // try to acquire a write lock + XplWaitOnLocalSemaphore(found_lock->access); + if (found_lock->readers == 0) { + // no readers - we're free + found_lock->writer = 1; + } else { + // need to queue for write access + NFairLockQueueEntry *ticket; + ticket = FairLockNewQueueEntry(found_lock, TRUE); + if (ticket) { + // since we're now in the queue, give up our access and wait our turn + XplSignalLocalSemaphore(found_lock->access); + XplWaitOnLocalSemaphore(ticket->signal); // wait til we reach the head of the queue + + XplWaitOnLocalSemaphore(found_lock->access); + + // now it's our turn, remove our ticket and try to get the write lock + FairLockRemoveQueueHead(found_lock); + if ((found_lock->readers > 0) || (found_lock->writer > 0)) { + // this is a bug + result = -3; + } + found_lock->writer = 1; + } else { + result = -2; + } + } + XplSignalLocalSemaphore(found_lock->access); + return result; +} + +/** + * Get a Shared lock to a store resource. A shared lock grants read-only + * access to the resource protected, so many threads may be accessing it + * simultaneously. If there is a queue already waiting, or the lock has + * been gained for Exclusive access, this function will cause the calling + * thread to wait in a queue for shared access to again become available. + * \bug Timeout simply doesn't work atm. + * \param client Store client who wishes to access a resource + * \param lock Where we want to save the lock we gain + * \param guid GUID of the store resource we want to access + * \param timeout How long we're willing to wait to gain the resource + * \return 0 if we successfully got the lock, an error number otherwise + */ +int +StoreGetSharedFairLockQuiet(StoreClient *client, NFairLock **lock, + uint64_t guid, time_t timeout) +{ + NFairLock *found_lock; + int result = 0; + + // first, find the lock + found_lock = FairLockFind(client->store, guid); + if (found_lock == NULL) return -1; // some error + *lock = found_lock; + + // next, try to acquire a shared lock + XplWaitOnLocalSemaphore(found_lock->access); + + if ((found_lock->writer == 0) && (found_lock->queue == NULL)) { + // no writers and no-one waiting, so we can freely read + found_lock->readers++; + } else { + // need to queue for read access + NFairLockQueueEntry *ticket; + ticket = FairLockNewQueueEntry(found_lock, FALSE); + if (ticket) { + // release our access and wait our turn + XplSignalLocalSemaphore(found_lock->access); + XplWaitOnLocalSemaphore(ticket->signal); + XplWaitOnLocalSemaphore(found_lock->access); + + FairLockRemoveQueueHead(found_lock); // remove our ticket + if (found_lock->writer > 0) { + // bug.. + result = -2; + } else { + found_lock->readers++; + // next person in queue might be a reader as well, + // so signal them + FairLockSignalQueueHead(found_lock); + } + } else { + // couldn't queue... + result = -1; + } + } + XplSignalLocalSemaphore(found_lock->access); + + // set the lock + *lock = found_lock; + + return result; +} + +/** + * Release a fair lock which was gained for read-only use. + * \param lock Reference to the lock we're giving up + * \return 0 if successful + */ +int +StoreReleaseSharedFairLockQuiet(NFairLock **lock) +{ + NFairLock *l = *lock; + + XplWaitOnLocalSemaphore(l->access); + l->readers--; + // signal any waiting threads + FairLockSignalQueueHead(l); + XplSignalLocalSemaphore(l->access); + *lock = NULL; + return 0; +} + +/** + * Release a fair lock which was gained for exclusive use. + * \param lock Reference to the lock we're giving up + * \return 0 if successful + */ +int +StoreReleaseExclusiveFairLockQuiet(NFairLock **lock) +{ + NFairLock *l = *lock; + + XplWaitOnLocalSemaphore(l->access); + l->writer = 0; + FairLockSignalQueueHead(l); + XplSignalLocalSemaphore(l->access); + *lock = NULL; + return 0; +} diff --git a/src/agents/store/lock.h b/src/agents/store/lock.h index 761de08..ee2d81b 100644 --- a/src/agents/store/lock.h +++ b/src/agents/store/lock.h @@ -27,6 +27,50 @@ XPL_BEGIN_C_LINKAGE #define STORE_COLLECTION_MAX_WATCHERS 8 /*********************** Structures ***********************/ + +// "fair" lock implementation + +#define SEM_UNLOCKED 1 +#define SEM_LOCKED 0 + +typedef struct _NFairLockQueueEntry { + // "signal" is used by the queueing process to wait its turn + XplSemaphore *signal; + struct _NFairLockQueueEntry *next; + BOOL is_writer; +} NFairLockQueueEntry; + +typedef struct _NFairLock { + char *lockname; + // "access" mediates changes to the internal data, shouldn't be held for long + XplSemaphore *access; + int readers; + int writer; + NFairLockQueueEntry *queue; +} NFairLock; + +typedef struct _NFairLockList { + struct _NFairLockList *next; + NFairLock *lock; +} NFairLockList; + +typedef struct _NFairLockPool { + // lock governs access to the Pool + XplSemaphore *lock; + NFairLockList *first; + NFairLockList *last; +} NFairLockPool; + +void StoreInitializeFairLocks(); +int StoreGetExclusiveFairLockQuiet(StoreClient *client, NFairLock **lock, + uint64_t guid, time_t timeout); +int StoreGetSharedFairLockQuiet(StoreClient *client, NFairLock **lock, + uint64_t guid, time_t timeout); +int StoreReleaseSharedFairLockQuiet(NFairLock **lock); +int StoreReleaseExclusiveFairLockQuiet(NFairLock **lock); + +// "normal" locks + typedef struct _NLock { /* FIXME: what's with all the unsigned longs? */ struct _NLock *nextHash; struct _NLock *prevHash; @@ -43,7 +87,7 @@ typedef struct _NLock { /* FIXME: what's with all the unsigned longs? */ /* used for WATCH command -- FIXME should be a linked list or growable array */ StoreClient *watchers[STORE_COLLECTION_MAX_WATCHERS]; - + NFairLock *fairlock; // HACK - gets around API changes for now. } NLockStruct; typedef struct { diff --git a/src/agents/store/store.c b/src/agents/store/store.c index dff4d6e..7767269 100644 --- a/src/agents/store/store.c +++ b/src/agents/store/store.c @@ -1264,17 +1264,25 @@ StoreReleaseSharedLock(StoreClient *client, NLockStruct *lock) /* gets an exclusive lock on the collection. */ +// FIXME: remove deprecated NLockStruct system CCode StoreGetCollectionLock(StoreClient *client, NLockStruct **lock, uint64_t coll) { - return StoreGetExclusiveLock(client, lock, coll, 3000); + int result; + printf("Want to lock " GUID_FMT " in %s\n", coll, client->storeName); + *lock = MemMalloc0(sizeof(NLockStruct)); + result = StoreGetExclusiveFairLockQuiet(client, &((*lock)->fairlock), coll, 3000); + if (result != 0) { + return ConnWriteStr(client->conn, MSG4120BOXLOCKED); + } + return 0; } CCode StoreReleaseCollectionLock(StoreClient *client, NLockStruct **lock) { - StoreReleaseExclusiveLock(client, *lock); - ReadNLockRelease(*lock); + StoreReleaseExclusiveFairLockQuiet(&((*lock)->fairlock)); + MemFree(*lock); *lock = NULL; } diff --git a/src/agents/store/stored.c b/src/agents/store/stored.c index 1d9baf7..91988e5 100644 --- a/src/agents/store/stored.c +++ b/src/agents/store/stored.c @@ -240,6 +240,9 @@ _XplServiceMain(int argc, char *argv[]) return -1; } + // create lock pool + StoreInitializeFairLocks(); + CONN_TRACE_INIT((char *)MsgGetWorkDir(NULL), "store"); CONN_TRACE_SET_FLAGS(CONN_TRACE_ALL); /* uncomment this line and pass '--enable-conntrace' to autogen to get the agent to trace all connections */ diff --git a/src/agents/store/stored.h b/src/agents/store/stored.h index 16ed883..048c2a8 100644 --- a/src/agents/store/stored.h +++ b/src/agents/store/stored.h @@ -22,6 +22,8 @@ #ifndef _STORED_H #define _STORED_H +#define LOGGERNAME "store" + #include XPL_BEGIN_C_LINKAGE @@ -36,7 +38,6 @@ XPL_BEGIN_C_LINKAGE #include #include -#define LOGGERNAME "store" #include /** Store stuff: **/