it IS possible to write to uncreated OIDs

src/server/core/core-write-op.cpp

@@ -36,11 +36,7 @@ static void write_op_exec_zero(void*, uint64_t, uint64_t);
 static void write_op_exec_omap_set(void*, char const* const*, char const* const*, const size_t*, size_t);
 static void write_op_exec_omap_rm_keys(void*, char const* const*, size_t);
 
-static oid_t lookup_oid(
-        sdskv_provider_handle_t ph,
-        sdskv_database_id_t name_db_id,
-        const char* object_name);
-static oid_t create_oid(
+static oid_t get_or_create_oid(
         sdskv_provider_handle_t ph,
         sdskv_database_id_t name_db_id,
         sdskv_database_id_t oid_db_id,
@@ -101,7 +97,7 @@ void write_op_exec_begin(void* u)
     sdskv_provider_handle_t sdskv_ph = vargs->srv_ctx->sdskv_ph;
     sdskv_database_id_t name_db_id = vargs->srv_ctx->name_db_id;
     sdskv_database_id_t oid_db_id  = vargs->srv_ctx->oid_db_id;
-    oid_t oid = lookup_oid(sdskv_ph, name_db_id, vargs->object_name);
+    oid_t oid = get_or_create_oid(sdskv_ph, name_db_id, oid_db_id, vargs->object_name);
     vargs->oid = oid;
 }
 
@@ -113,23 +109,13 @@ void write_op_exec_end(void* u)
 void write_op_exec_create(void* u, int exclusive)
 {
     ENTERING;
-	auto vargs = static_cast<server_visitor_args_t>(u);
-    sdskv_provider_handle_t sdskv_ph = vargs->srv_ctx->sdskv_ph;
-    sdskv_database_id_t name_db_id = vargs->srv_ctx->name_db_id;
-    sdskv_database_id_t oid_db_id  = vargs->srv_ctx->oid_db_id;
-    const char *object_name = vargs->object_name;
+    auto vargs = static_cast<server_visitor_args_t>(u);
     oid_t oid = vargs->oid; 
     if(oid == 0) {
-        /* create OID if it didn't exist when this write op began */
-        oid = create_oid(sdskv_ph, name_db_id, oid_db_id, object_name);
-        if(oid == 0) {
-            /* XXX try one last time to lookup to deal with race to create OID */
-            oid = lookup_oid(sdskv_ph, name_db_id, object_name);
-            if(oid == 0) {
-                ERROR fprintf(stderr,"could not create OID for %s!\n", object_name); 
-            }
-        }
+        ERROR fprintf(stderr,"oid == 0\n");
     }
+    /* nothing to do, the object is actually created in write_op_exec_begin
+       if it did not exist before */
     LEAVING;
 }
 
@@ -545,25 +531,7 @@ void write_op_exec_omap_rm_keys(void* u, char const* const* keys, size_t num_key
     LEAVING;
 }
 
-static oid_t lookup_oid(
-        sdskv_provider_handle_t ph,
-        sdskv_database_id_t name_db_id,
-        const char* object_name)
-{
-    ENTERING;
-    LEAVING;
-    oid_t oid = 0;
-    hg_size_t vsize = sizeof(oid);
-    int ret;
-    ret = sdskv_get(ph, name_db_id, (const void*)object_name,
-            strlen(object_name)+1, &oid, &vsize);
-    if(SDSKV_SUCCESS == ret)
-        return oid;
-    else
-        return 0;
-}
-
-static oid_t create_oid(
+static oid_t get_or_create_oid(
         sdskv_provider_handle_t ph,
         sdskv_database_id_t name_db_id,
         sdskv_database_id_t oid_db_id,
@@ -571,53 +539,60 @@ static oid_t create_oid(
 {
     ENTERING;
     oid_t oid = 0;
+    hg_size_t s;
     int ret;
-    std::hash<std::string> hash_fn;
-    oid = hash_fn(std::string(object_name));
-    hg_size_t s = strlen(object_name)+1;
-    char *name_check = (char *)malloc(s);
-    if(!name_check) {
-        LEAVING;
-        return 0;
-    }
-    while(1) {
-        /* avoid hash collisions by checking this oid mapping */
-        ret = sdskv_get(ph, oid_db_id, (const void*)&oid,
-            sizeof(oid), (void *)name_check, &s);
-        if(ret == SDSKV_SUCCESS) {
-            if(strncmp(object_name, name_check, s) == 0) {
-                /* the object has been created by someone else in the meantime...  */
-                free(name_check);
-                LEAVING;
-                return oid;
+
+    s = sizeof(oid);
+    ret = sdskv_get(ph, name_db_id, (const void*)object_name,
+            strlen(object_name)+1, &oid, &s);
+    if(SDSKV_ERR_UNKNOWN_KEY == ret) {
+        std::hash<std::string> hash_fn;
+        oid = hash_fn(std::string(object_name));
+        s = strlen(object_name)+1;
+        char *name_check = (char *)malloc(s);
+        if(!name_check) {
+            LEAVING;
+            return 0;
+        }
+        while(1) {
+            /* avoid hash collisions by checking this oid mapping */
+            ret = sdskv_get(ph, oid_db_id, (const void*)&oid,
+                sizeof(oid), (void *)name_check, &s);
+            if(ret == SDSKV_SUCCESS) {
+                if(strncmp(object_name, name_check, s) == 0) {
+                    /* the object has been created by someone else in the meantime...  */
+                    free(name_check);
+                    LEAVING;
+                    return oid;
+                }
+                oid++;
+                continue;
             }
-            oid++;
-            continue;
+            break;
+        }
+        free(name_check);
+        // we make sure we stopped at an unknown key (not another SDSKV error)
+        if(ret != SDSKV_ERR_UNKNOWN_KEY) {
+            fprintf(stderr, "[ERROR] ret != SDSKV_ERR_UNKNOWN_KEY (ret = %d)\n", ret);
+            LEAVING;
+            return 0;
+        }
+        // set name => oid
+        ret = sdskv_put(ph, name_db_id, (const void*)object_name,
+                strlen(object_name)+1, &oid, sizeof(oid));
+        if(ret != SDSKV_SUCCESS) {
+            fprintf(stderr, "[WARNING] after sdskv_put(name->oid), ret != SDSKV_SUCCESS (ret = %d)\n", ret);
+            LEAVING;
+            return 0;
+        }
+        // set oid => name
+        ret = sdskv_put(ph, oid_db_id, &oid, sizeof(oid),
+                (const void*)object_name, strlen(object_name)+1);
+        if(ret != SDSKV_SUCCESS) {
+            fprintf(stderr, "[WARNING] after sdskv_put(oid->name), ret != SDSKV_SUCCESS (ret = %d)\n", ret);
+            LEAVING;
+            return 0;
         }
-        break;
-    }
-    free(name_check);
-    // we make sure we stopped at an unknown key (not another SDSKV error)
-    if(ret != SDSKV_ERR_UNKNOWN_KEY) {
-        fprintf(stderr, "[ERROR] ret != SDSKV_ERR_UNKNOWN_KEY (ret = %d)\n", ret);
-        LEAVING;
-        return 0;
-    }
-    // set name => oid
-    ret = sdskv_put(ph, name_db_id, (const void*)object_name,
-            strlen(object_name)+1, &oid, sizeof(oid));
-    if(ret != SDSKV_SUCCESS) {
-        fprintf(stderr, "[WARNING] after sdskv_put(name->oid), ret != SDSKV_SUCCESS (ret = %d)\n", ret);
-        LEAVING;
-        return 0;
-    }
-    // set oid => name
-    ret = sdskv_put(ph, oid_db_id, &oid, sizeof(oid),
-            (const void*)object_name, strlen(object_name)+1);
-    if(ret != SDSKV_SUCCESS) {
-        fprintf(stderr, "[WARNING] after sdskv_put(oid->name), ret != SDSKV_SUCCESS (ret = %d)\n", ret);
-        LEAVING;
-        return 0;
     }
     LEAVING;
     return oid;