First declare/provide/withdraw/dispose roundtrip

.gitignore

@@ -24,3 +24,4 @@ maestro-core-*.tar.gz
 TAGS
 tags
 /tests/check_conductor
+/tests/check_pool_local

include/Makefile.am

@@ -68,7 +68,8 @@ noinst_HEADERS = \
 	maestro/i_utstack.h    \
 	maestro/i_drc.h        \
 	maestro/i_tpl.h        \
-	maestro/i_conductor_protocol.h 
+	maestro/i_conductor_protocol.h \
+	maestro/i_pool.h       
 
 
 

include/maestro/i_cdo.h

@@ -76,6 +76,11 @@ typedef uint32_t mstro_cdo_state;
                                    | MSTRO_CDO_STATE_DEMANDED   \
                                    | MSTRO_CDO_STATE_REVOKED)
 
+/** CDO that is withdrawable */
+#define MSTRO_CDO_STATE_DISPOSABLE (  MSTRO_CDO_STATE_DECLARED  \
+                                    | MSTRO_CDO_STATE_WITHDRAWN \
+                                    | MSTRO_CDO_STATE_REVOKED)
+  
 /** CDO dead */
 #define MSTRO_CDO_STATE_DEAD      (1<<7)
 
@@ -113,6 +118,20 @@ struct mstro_cdo_id {
 mstro_status
 mstro_cdo_id__str(const struct mstro_cdo_id *cdoid, char **name);
 
+/**@brief Provide a convenient scope in which VARNAME contains the CDO
+ * ID as a string */
+#define WITH_CDO_ID_STR(varname, cdoidptr, body)                          \
+  do {                                                                  \
+    char *varname = alloca(MSTRO_CDO_ID_STR_LEN);                       \
+    if(MSTRO_OK!=mstro_cdo_id__str(cdoidptr, &varname)) {               \
+      ERR("Failed to construct printable CDO ID for %p\n", cdoidptr);   \
+    } else {                                                            \
+      body                                                              \
+          }                                                             \
+  } while(0)                                                            
+
+
+
 /** produce an integral hash for CDOID in *hash. */
 static inline
 mstro_status

include/maestro/i_globals.h

@@ -44,4 +44,15 @@ extern struct mstro_core_initdata * g_initdata;
 /** lock to protect @ref g_initdata */
 extern pthread_mutex_t g_initdata_mtx;
 
+
+/** Hash table of local CDO handles. Protected by g_mstro_pool_local_mtx */
+extern struct mstro_pool_entry * g_mstro_pool_local;
+
+/** Lock protecting @ref g_mstro_pool_local */
+extern pthread_mutex_t g_mstro_pool_local_mtx;
+
+
+/** The NULL CDO-ID */
+extern struct mstro_cdo_id MSTRO_CDO_ID_NULL;
+
 #endif /* MAESTRO_I_GLOBALS_H_ */

include/maestro/i_pool.h

+/* -*- mode:c -*- */
+/** @file
+ ** @brief Maestro Pool functions, local and global
+ **/
+/*
+ * Copyright (C) 2019 Cray Computer GmbH
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * 3. Neither the name of the copyright holder nor the names of its
+ *    contributors may be used to endorse or promote products derived from
+ *    this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+ * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+ * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef MAESTRO_I_POOL_H_
+#define MAESTRO_I_POOL_H_ 1
+
+#include "maestro/i_state.h"
+#include "maestro/i_cdo.h"
+#include "maestro/i_uthash.h"
+#include <pthread.h>
+
+
+/** Add CDO to local pool, either in a fresh entry or add to existing one */
+mstro_status
+mstro_pool__add(mstro_cdo cdo);
+
+/** Remove CDO from the local pool */
+mstro_status
+mstro_pool__remove(mstro_cdo cdo);
+
+
+/**@brief Initialize pool
+ *
+ * Initializes the local part of the pool, and notifies conductor of it
+ */
+mstro_status
+mstro_pool_init(void);
+
+/**@brief Finalize pool
+ *
+ * Shuts down local pool functionality.
+ *
+ * Notifies conductor to possibly pull CDOs that are still needed off
+ * the local resources, then shuts down local pool operations.
+ */
+mstro_status
+mstro_pool_finalize(void);
+
+#endif /* MAESTRO_I_GLOBALS_H_ */

maestro-valgrind.supp

+# suppressions file to be passed to valgrind with
+#   --suppressions=/path/to/maestro-valgrind.supp
+
+# the localhost name string is not ever deallocated
+{
+  g_localhost_leak
+  Memcheck:Leak
+  fun:malloc
+  fun:mstro_location_aware_log 
+  fun:mstro_init
+}
+

maestro/Makefile.am

@@ -43,7 +43,8 @@ libmaestro_core_la_SOURCES = \
 	status.c \
 	uuid.c uuid_sha1.c uuid_str.c uuid_ui128.c uuid_ui64.c base64.c\
 	drc.c \
-	env.c logging.c tpl.c 
+	env.c logging.c tpl.c \
+        pool.c
 
 # Conductor implementation varies depending on configuration
 if WITH_OFI_CONDUCTOR

maestro/cdo.c

@@ -34,6 +34,8 @@
  */
 
 #include "maestro/i_cdo.h"
+#include "maestro/logging.h"
+#include "maestro/i_pool.h"
 
 #include <string.h>
 #include <assert.h>
@@ -50,6 +52,7 @@ static inline
 mstro_status
 mstro_cdo__free(mstro_cdo *cdoptr)
 {
+  DEBUG("cdoptr %p, cdo %p, name %s\n", cdoptr, *cdoptr, (*cdoptr)->name);
   assert(cdoptr!=NULL);
   assert(*cdoptr!=NULL);
   if((*cdoptr)->name)
@@ -78,19 +81,28 @@ mstro_cdo_declare(const unsigned char *name,
                   mstro_cdo_decl_attr attributes,
 		  mstro_cdo *result)
 {
-  if(name==NULL)
+  if(name==NULL) {
+    ERR("name cannot be NULL\n");
     return MSTRO_INVARG;
-  if(result==NULL)
+  }
+  if(result==NULL) {
+    ERR("declaration handle storage cannot be NULL\n");
     return MSTRO_INVOUT;
+  }
 
   *result = mstro_cdo__alloc();
-  if(*result==NULL)
+  if(*result==NULL) {
+    ERR("Cannot allocate for CDO declaration\n");
     return MSTRO_NOMEM;
+  }
 
   size_t len = strlen((char*)name);
   (*result)->name = malloc(sizeof(unsigned char)*(len+1));
-  if((*result)->name==NULL) 
+  if((*result)->name==NULL)  {
+    ERR("Cannot allocate for CDO name\n");
     mstro_cdo__free(result);
+    return MSTRO_NOMEM;
+  }
   memcpy((*result)->name, name, len+1);
   
   (*result)->state = MSTRO_CDO_STATE_CREATED;
@@ -101,10 +113,16 @@ mstro_cdo_declare(const unsigned char *name,
   mstro_status s = mstro_cdo_id_from_name(name, &(*result)->id);
 
   if(s!=MSTRO_OK) {
+    ERR("Failed to compute local ID for CDO");
     mstro_cdo__free(result);
     return s;
   }
   
+  WITH_CDO_ID_STR(idstr,&(*result)->id,
+                  INFO("Declared CDO `%s', (local ID: %s)\n",
+                       (*result)->name, idstr);
+                  );
+ 
   return MSTRO_OK; 
 }
 
@@ -118,15 +136,104 @@ mstro_cdo_attribute_set(mstro_cdo cdo, const char*key, void* val)
   return MSTRO_UNIMPL;
 }
 
+mstro_status
+mstro_cdo_declaration_seal(mstro_cdo cdo)
+{
+  if(cdo==0) {
+    ERR("NULL CDO illegal\n");
+    return MSTRO_INVARG;
+  }
+  if(cdo->state !=MSTRO_CDO_STATE_CREATED) {
+    ERR("CDO `%s' has state %d, cannot seal\n", cdo->name, cdo->state);
+    return MSTRO_INVARG;
+  }
+  WARN("Missing conductor communication to confirm CDO-ID\n");
+  cdo->gid = cdo->id;
+  cdo->state = MSTRO_CDO_STATE_DECLARED;
+
+  return MSTRO_OK;
+}
+
 mstro_status
 mstro_cdo_provide(mstro_cdo cdo)
 {
+  mstro_status status;
+  if(cdo==NULL) {
+    ERR("NULL CDO invalid\n");
+    return MSTRO_INVARG;
+  }
+  
+  if(cdo->state!=MSTRO_CDO_STATE_DECLARED) {
+    if(cdo->state==MSTRO_CDO_STATE_CREATED) {
+      INFO("Auto-sealing CDO `%s'\n", cdo->name);
+      status = mstro_cdo_declaration_seal(cdo);
+      if(status!=MSTRO_OK) {
+        return status;
+      }
+    } else {
+      ERR("CDO `%s' state %d illegal for PROVIDE operation\n",
+          cdo->name, cdo->state);
+      return MSTRO_INVARG;
+    }
+  }
+
+  /* We have a sealed CDO. Good. */
+  WITH_CDO_ID_STR(
+      id, &cdo->gid,
+      {
+        status = mstro_pool__add(cdo);
+        
+        if(status!=MSTRO_OK) {
+          ERR("Failed to add CDO %s (id %s) to pool\n", cdo->name, id);
+          return status;
+        } else {
+          cdo->state = MSTRO_CDO_STATE_PROVIDED;
+          INFO("Added CDO `%s' (id %s) to pool\n",
+               cdo->name, id);
+          return MSTRO_OK;
+          
+        }
+      });
+  /* not reached */
   return MSTRO_UNIMPL;
 }
 
 mstro_status
 mstro_cdo_withdraw(mstro_cdo cdo)
 {
+  if(cdo==NULL) {
+    ERR("NULL CDO invalid\n");
+    return MSTRO_INVARG;
+  }
+  WITH_CDO_ID_STR(idstr, &cdo->gid, {
+      DEBUG("Removing CDO `%s' (id %s) from pool\n",
+            cdo->name, idstr);});
+
+  if(! (cdo->state & MSTRO_CDO_STATE_POOLED)) {
+    WITH_CDO_ID_STR(
+        idstr, &cdo->gid,
+        ERR("CDO `%s' (id %s) not pooled, cannot remove it\n",
+            cdo->name, idstr););
+    return MSTRO_INVARG;
+  }
+
+  mstro_status status = mstro_pool__remove(cdo);
+
+  if(status!=MSTRO_OK) {
+    WITH_CDO_ID_STR(
+        idstr, &cdo->gid,
+        ERR("Failed to remove CDO %s (id %s) from pool\n",
+            cdo->name, idstr););
+    return status;
+  } else {
+    cdo->state = MSTRO_CDO_STATE_WITHDRAWN;
+    WITH_CDO_ID_STR(
+        idstr, &cdo->gid,
+        INFO("Withdrew CDO `%s' (id %s) from pool\n",
+             cdo->name, idstr););
+    return MSTRO_OK;
+  }
+  
   return MSTRO_UNIMPL;
 }
 
@@ -151,6 +258,39 @@ mstro_cdo_revoke(mstro_cdo cdo)
 mstro_status
 mstro_cdo_dispose(mstro_cdo cdo)
 {
+  if(cdo==NULL) {
+    ERR("NULL CDO invalid\n");
+    return MSTRO_INVARG;
+  }
+
+  if(! (cdo->state & MSTRO_CDO_STATE_DISPOSABLE)) {
+    WITH_CDO_ID_STR(
+        idstr, &cdo->gid,
+        ERR("CDO `%s' (id %s) state %d, not disposable\n",
+            cdo->name, idstr, cdo->state););
+    return MSTRO_INVARG;
+  }
+
+  /* Careful! We need can only get name, id, etc. before the disposal
+   * happens.
+   * FIXME: This should actually be #ifdef-guarded depending on log
+   * level to save overhead for high CDO disposal rate situations. */
+  WITH_CDO_ID_STR(
+      idstr, &cdo->gid,
+      {
+        INFO("Disposing CDO `%s' (id %s)\n",
+             cdo->name, idstr);
+        
+        mstro_status status = mstro_cdo__free(&cdo);
+        if(status!=MSTRO_OK) {
+          ERR("Failed to deallocate CDO `%s' (id %s)\n",
+              cdo->name, idstr);
+          return status;
+        } else {
+          return MSTRO_OK;
+        }
+      });
+  
   return MSTRO_UNIMPL;
 }
 

maestro/core.c

@@ -3,6 +3,7 @@
 #include "maestro/logging.h"
 
 #include "maestro/i_globals.h"
+#include "maestro/i_pool.h"
 
 #include <pthread.h>
 #include <stdio.h>
@@ -87,6 +88,10 @@ mstro_core_init(const char *workflow_name,
   pthread_mutex_unlock(&g_initdata_mtx);
 
   
+  /* Start local pool */
+  status = mstro_pool_init();
+  
+
   return status;
 }
 
@@ -94,6 +99,12 @@ mstro_status
 mstro_core_finalize(void)
 {
   mstro_status status = MSTRO_UNIMPL;
+
+  status = mstro_pool_finalize();
+  if(status!=MSTRO_OK) {
+    ERR("Pool finalization failed: %d\n", status);
+    goto BAILOUT;
+  }
   pthread_mutex_lock(&g_initdata_mtx);
   if(g_initdata==NULL) {
     /* in case someone calls finalize before init: */
@@ -114,6 +125,7 @@ mstro_core_finalize(void)
   }
   pthread_mutex_unlock(&g_initdata_mtx);
 
+BAILOUT:  
   return status;
 }
 

maestro/globals.c

 /* Global variable instantiation */
 #include "maestro/i_globals.h"
+#include "maestro/i_cdo.h"
 
 /** Init data; set at mstro_core_init() time;
     protected by @ref g_initdata_mtx */
 struct mstro_core_initdata * g_initdata = NULL;
 /** lock to protect @ref g_initdata */
 pthread_mutex_t g_initdata_mtx = PTHREAD_MUTEX_INITIALIZER;
+
+
+
+pthread_mutex_t g_mstro_pool_local_mtx = PTHREAD_MUTEX_INITIALIZER;
+struct mstro_pool_entry * g_mstro_pool_local = NULL;
+
+
+struct mstro_cdo_id MSTRO_CDO_ID_NULL = MSTRO_CDO_ID_INVALID;

maestro/init.c

@@ -35,13 +35,22 @@
  */
 #include "maestro.h"
 #include "maestro/core.h"
+#include "maestro/logging.h"
 
 mstro_status
 mstro_init(const char *workflow_name,
            const char *component_name,
            uint64_t component_index)
 {
-  return mstro_core_init(workflow_name, component_name, component_index);
+  mstro_status status = MSTRO_UNIMPL;
+  /* FIXME: read config */
+  WARN("should read config file here\n");
+
+  status = mstro_core_init(workflow_name, component_name, component_index
+                           /* FIXME: core config file fragment parameter */
+                           );
+
+  return status;
 }
 
 

maestro/logging.c

@@ -87,6 +87,7 @@ const char *
 mstro_local_hostname(void)
 {
   if(g_localhost==NULL) {
+    /* valgrind users: This will intentionally leak. */
     g_localhost = malloc(256*sizeof(char));
     if(g_localhost==NULL) {
       g_localhost = "[unknown]";

maestro/pool.c

+#include "maestro/i_pool.h"
+#include "maestro/logging.h"
+#include "maestro/i_cdo.h"
+#include "maestro/i_globals.h"
+#include <pthread.h>
+#include <assert.h>
+
+#define NUM_CDOS_PER_POOL_ENTRY 4
+/** The local pool is made up of these entries.
+ **
+ ** It maps (globally valid) CDO IDs to the set of all local handles.
+ **
+ ** Waiting on changes of an entry permits to detect availabilty of
+ ** matching src/dst pairs of CDOs.
+ **
+ ** 
+ **/
+struct mstro_pool_entry {
+  UT_hash_handle hh;           /**< make structure hashable */
+  struct mstro_cdo_id   cdoid; /**< primary key */
+  pthread_mutex_t mtx;         /**< lock for (the handles of) this entry */
+  pthread_cond_t  cvar;        /**< condition variable to wait for entry changes */
+  size_t         num_cdo_handles; /**< number of CDO handles for this ID */
+  size_t         num_free_slots;  /**< number of free cdo_handles slots */
+  mstro_cdo      *cdo_handles;    /**< the CDO handles */
+};
+
+/** utility to lock a critical section for a pool entry */
+
+#define WITH_LOCKED_POOL_ENTRY(entry, errorlabel, body)                 \
+  do {                                                                  \
+    if(pthread_mutex_lock(&e->mtx)!=0) {                                \
+      WITH_CDO_ID_STR(                                                  \
+          idstr, & ((e)->cdoid), {                                      \
+            ERR("Failed to lock mutex for pool entry %p (id %s)\n",     \
+                e, idstr);                                              \
+          });                                                           \
+      goto errorlabel;                                                  \
+    }                                                                   \
+    do { body } while(0);                                               \
+    if(pthread_mutex_unlock(&e->mtx)!=0) {                              \
+      WITH_CDO_ID_STR(                                                  \
+          idstr, & ((e)->cdoid), {                                      \
+            ERR("Failed to unlock mutex for pool entry %p (id %s)\n",   \
+                e, idstr);                                              \
+          });                                                           \
+      goto errorlabel;                                                  \
+    }                                                                   \
+  } while(0)
+
+/** Allocate pool entry */
+static inline
+mstro_status
+mstro_pool_entry__alloc(struct mstro_pool_entry **res)
+{
+  if(res==NULL) {
+    ERR("NULL pool entry handle destination\n");
+    return MSTRO_INVARG;
+  }
+
+  mstro_status status;
+  
+  int s1,s2;
+  *res = calloc(1,sizeof(struct mstro_pool_entry));
+  if(*res==NULL) {
+    ERR("Cannot allocate pool entry\n");
+    return MSTRO_NOMEM;
+  } else {
+    s1 = pthread_mutex_init(&((*res)->mtx), NULL);
+    s2 = pthread_cond_init(&((*res)->cvar), NULL);
+    (*res)->num_free_slots = 0;
+    (*res)->num_cdo_handles = 0;
+    (*res)->cdo_handles = NULL;
+    if(s1!=0 || s2!=0) {
+      ERR("pthread mutex or cond init failed\n");
+      status = MSTRO_FAIL;
+      goto BAILOUT;
+    }
+  }
+  status = MSTRO_OK;
+BAILOUT:
+  if(status!=MSTRO_OK) {
+    if(*res!=NULL) {
+      if(s1==0) {
+        pthread_mutex_destroy(&((*res)->mtx));
+      }
+      if(s2==0) {
+        pthread_cond_destroy(&((*res)->cvar));
+      }
+      if((*res)->cdo_handles!=NULL) {
+        free((*res)->cdo_handles);
+      }
+      free(*res);
+      *res=NULL;
+    }
+  }
+  return status;
+}
+
+/** Destroy a pool entry */
+static inline
+mstro_status
+mstro_pool_entry__destroy(struct mstro_pool_entry *e)
+{
+  mstro_status status = MSTRO_OK;
+  if(e==NULL)
+    return MSTRO_INVARG;
+  int s;
+  s=pthread_mutex_destroy(&e->mtx);
+  if(s!=0) {
+    ERR("Failed to destroy mutex\n");
+    status = MSTRO_FAIL;
+  }
+  s=pthread_cond_destroy(&e->cvar);
+  if(s!=0) {
+    ERR("Failed to destroy cvar\n");
+    status = MSTRO_FAIL;
+  }
+
+  free(e->cdo_handles);
+  free(e);
+
+  return status;
+}
+
+
+/** Add CDO to pool entry */
+static inline
+mstro_status
+mstro_pool_entry__add(struct mstro_pool_entry *e, mstro_cdo cdo)
+{
+  mstro_status status = MSTRO_OK;
+  int s;
+  
+  if(e==NULL || cdo==NULL)
+    return MSTRO_INVARG;
+
+  WITH_LOCKED_POOL_ENTRY(
+      e, BAILOUT_ERR, {
+        if(! mstro_cdo_id__equal(e->cdoid, cdo->gid)) {
+          ERR("Trying to add CDO to wrong pool entry (or gid not ready)\n");
+          status = MSTRO_INVARG;
+        } else {
+          if(e->num_free_slots==0) {
+            mstro_cdo * tmp = realloc(e->cdo_handles,
+                                      sizeof(mstro_cdo *)
+                                      *(e->num_cdo_handles
+                                        + NUM_CDOS_PER_POOL_ENTRY));
+            if(tmp==NULL) {
+              ERR("Failed to re-allocate handles in pool entry\n");
+              status = MSTRO_NOMEM;
+            } else {
+              e->cdo_handles = tmp;
+              e->num_free_slots += NUM_CDOS_PER_POOL_ENTRY;
+              
+              e->cdo_handles[e->num_cdo_handles] = cdo;
+              e->num_cdo_handles++;
+              e->num_free_slots--;
+            }
+          }
+        }
+      });
+
+  
+BAILOUT:
+  return status;
+BAILOUT_ERR:
+  status = MSTRO_FAIL;
+  goto BAILOUT;
+}
+
+
+mstro_status
+mstro_pool__add(mstro_cdo cdo)
+{
+  mstro_status status = MSTRO_OK;
+  int s;
+  if(cdo==NULL) {
+    return MSTRO_INVARG;
+  }
+
+  if(mstro_cdo_id__equal(MSTRO_CDO_ID_NULL, cdo->gid)) {
+    ERR("CDO has no gid\n");
+    return MSTRO_INVARG;
+  }
+
+  /* find in pool */
+  WITH_CDO_ID_STR(id, &cdo->gid, {
+      INFO("Adding CDO `%s' (id %s) to pool\n", cdo->name, id);
+    });
+
+  s = pthread_mutex_lock(&g_mstro_pool_local_mtx);
+  if(s!=0) {
+    ERR("Failed to lock the pool mutex\n");
+    status = MSTRO_FAIL;
+    goto BAILOUT;
+  }
+  struct mstro_pool_entry *e=NULL;
+  DEBUG("finding entry in pool\n");
+  HASH_FIND(hh, g_mstro_pool_local,
+            &(cdo->gid), sizeof(struct mstro_cdo_id),
+            e);
+  if(e==NULL) {
+    DEBUG("not found, creating new one\n");
+    /* create new (empty) entry, add to table */
+    status = mstro_pool_entry__alloc(&e);
+    if(status!=MSTRO_OK) {
+      ERR("Failed to allocate pool entry\n");
+      goto BAILOUT_UNLOCK;
+    }
+    e->cdoid = cdo->gid;
+    DEBUG("adding new entry to pool\n");
+    HASH_ADD(hh, g_mstro_pool_local,
+             cdoid, sizeof(struct mstro_cdo_id),
+             e);
+  }
+
+  status = mstro_pool_entry__add(e,cdo);
+  if(status!=MSTRO_OK) {
+    ERR("Failed to add CDO to pool entry\n");
+  } else {
+    DEBUG("added CDO to pool entry\n");
+  }
+
+  /* release lock */
+BAILOUT_UNLOCK:
+  s = pthread_mutex_unlock(&g_mstro_pool_local_mtx);
+  if(s!=0) {
+    status = MSTRO_FAIL;
+    goto BAILOUT;
+  } else {
+    assert(status==MSTRO_OK);
+    status = MSTRO_OK;
+  }
+
+BAILOUT:
+  return status;
+}
+
+mstro_status
+mstro_pool__remove(mstro_cdo cdo)
+{
+  if(cdo==NULL) {
+    ERR("NULL CDO invalid\n");
+    return MSTRO_INVARG;
+  }
+
+  /* find entry */
+  mstro_status status = MSTRO_UNIMPL;
+  
+  int s = pthread_mutex_lock(&g_mstro_pool_local_mtx);
+  if(s!=0) {
+    ERR("Failed to lock pool\n");
+    return MSTRO_FAIL;
+  }
+
+  struct mstro_pool_entry *e=NULL;
+  HASH_FIND(hh, g_mstro_pool_local,
+            &(cdo->gid), sizeof(struct mstro_cdo_id),
+            e);
+  if(e==NULL) {
+    WITH_CDO_ID_STR(
+        idstr, &cdo->gid,
+            ERR("Pooled CDO %s (id %s) has no pool entry !?\n",
+                cdo->name, idstr););
+    status = MSTRO_FAIL;
+  } else {
+    /* find it among handles */
+    WITH_LOCKED_POOL_ENTRY(
+        e, BAILOUT_ERR, {
+          size_t i;
+          for(i=0; i<e->num_cdo_handles; i++) {
+            if(e->cdo_handles[i]==cdo)
+              break;
+          }
+          if(i==e->num_cdo_handles) {
+            WITH_CDO_ID_STR(
+                idstr, &cdo->gid,
+                ERR("Pooled CDO %s (id %s) not present in its pool entry !?\n",
+                    cdo->name, idstr););
+            status = MSTRO_FAIL;
+          } else {
+            /* got it at index i */
+            e->num_cdo_handles--;
+            e->num_free_slots++;
+            
+            if(e->num_cdo_handles>0) {
+              /* swap last one into the place 'i' */
+              e->cdo_handles[i] = e->cdo_handles[e->num_cdo_handles];
+            }
+            status = MSTRO_OK;
+          }
+        });
+  }
+    
+BAILOUT_UNLOCK_POOL:
+  s = pthread_mutex_unlock(&g_mstro_pool_local_mtx);
+  if(s!=0) {
+    ERR("Failed to unlock pool\n");
+    status = MSTRO_FAIL;
+    goto BAILOUT;
+  }
+
+BAILOUT:
+  return status;
+
+BAILOUT_ERR:
+  status = MSTRO_FAIL;
+  goto BAILOUT_UNLOCK_POOL;
+}
+
+
+
+mstro_status
+mstro_pool_init(void)
+{
+  INFO("Initializing pool\n");
+  /* FIXME: needs conductor communications */
+  return MSTRO_OK;
+}
+
+mstro_status
+mstro_pool_finalize(void)
+{
+  INFO("Cleaning up pool\n");
+  mstro_status status = MSTRO_OK;
+  
+  int s = pthread_mutex_lock(&g_mstro_pool_local_mtx);
+  if(s!=0) {
+    status = MSTRO_FAIL;
+    goto BAILOUT;
+  }
+  size_t num_entries = HASH_COUNT(g_mstro_pool_local);
+  INFO("Pool has %zu entries\n", num_entries);
+  if(num_entries>0) {
+    WARN("NOT signaling conductor to syphon off pool entries\n");
+    struct mstro_pool_entry *e,*tmp;
+    HASH_ITER(hh, g_mstro_pool_local, e, tmp) {
+      HASH_DEL(g_mstro_pool_local, e);
+      if(e->num_cdo_handles>0) {
+        WITH_CDO_ID_STR(id,&e->cdoid,{
+            WARN("Pool entry for id %s has %d CDOs; if any process is still blocking on this it will suffer now\n",
+                 id, e->num_cdo_handles);});
+      }
+
+      status = mstro_pool_entry__destroy(e);
+    }
+  }
+  g_mstro_pool_local = NULL;
+  s = pthread_mutex_unlock(&g_mstro_pool_local_mtx);
+  if(s!=0)
+    status = (status==MSTRO_OK) ? MSTRO_FAIL : status;
+
+BAILOUT:
+  return status;
+}

tests/Makefile.am

@@ -37,11 +37,13 @@ LDADD = $(top_builddir)/libmaestro.la
 check_HEADERS = cheat.h
 
 TESTS = check_version check_init\
-        check_declare
+        check_declare \
+	check_pool_local 
 #        check_conductor
 
 check_PROGRAMS = check_version check_init \
                  check_declare \
+                 check_pool_local \
                  check_conductor
 
 check_SCRIPTS = 

tests/check_pool_local.c

+/* check CDO declare */
+/*
+ * Copyright (C) 2019 Cray Computer GmbH
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * 3. Neither the name of the copyright holder nor the names of its
+ *    contributors may be used to endorse or promote products derived from
+ *    this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+ * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+ * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+/* needed before inclusion of cheat.h: */
+#ifndef __BASE_FILE__
+#define __BASE_FILE__ __FILE__
+#endif
+
+#include "cheat.h"
+
+#include "maestro.h"
+#include <string.h>
+
+
+/* this tests that we can call toplevel API init/finalize  */
+CHEAT_TEST(cdo_local_pool_works,
+           cheat_assert(MSTRO_OK == mstro_init("Tests","POOL",0));
+           unsigned char name[] = "Test CDO name üÜöÖäÄ";
+           mstro_cdo cdo=NULL;
+           cheat_assert(MSTRO_OK == mstro_cdo_declare(name, MSTRO_ATTR_DEFAULT, &cdo));
+           cheat_yield();
+           cheat_assert(MSTRO_OK == mstro_cdo_provide(cdo));
+           cheat_yield();
+           cheat_assert(MSTRO_OK == mstro_cdo_withdraw(cdo));
+           cheat_assert(MSTRO_OK == mstro_cdo_dispose(cdo));
+           cheat_assert(MSTRO_OK == mstro_finalize());
+           )
+
+
+