forb: Split forb_port_destroy() to stop and destroy phases

[frescor/forb.git] / src / forb.c

diff --git a/src/forb.c b/src/forb.c
index 63e51eb72979063cfb3fb02a4001d9dc6a97cee7..309c6602e8f8b5a84b469be748058385e5b76e5e 100644 (file)
--- a/src/forb.c
+++ b/src/forb.c
@@ -150,9 +150,8 @@ forb_init_tmp_dir(void)
 static void *
 forb_execution_thread(void *arg)
 {
-       forb_orb orb = arg;
-
-       forb_execute_object(orb);
+       forb_executor_t *executor = arg;
+       forb_executor_run(executor);
        return NULL;
 }
 
@@ -236,10 +235,15 @@ forb_init(int *argc, char **argv[], const struct forb_init_attr *attr)
         * can accept remote request to our new ORB. */
        forb_objects_nolock_insert(forb, orb);
 
+       ret = forb_executor_init(&forb->default_executor);
+       if (ret) goto err2;
+       ret = forb_executor_register_object(&forb->default_executor, orb);
+       if (ret) goto err3;
+
        ret = fosa_thread_create(&forb->execution_thread, NULL,
-                                forb_execution_thread, orb);
+                                forb_execution_thread, &forb->default_executor);
        if (ret != 0)
-               goto err2;
+               goto err3;
 
        /* FIXME: I do not know how to deregister the exit handler if
         * forb_destroy() is called manually. */
@@ -258,7 +262,7 @@ forb_init(int *argc, char **argv[], const struct forb_init_attr *attr)
                }
        err_free_unix:
                free(port);
-               goto err2;
+               goto err3;
        unix_ok:;
        }
 #endif
@@ -279,13 +283,14 @@ forb_init(int *argc, char **argv[], const struct forb_init_attr *attr)
                }
        err_free_inet:
                free(port);
-               goto err2;
+               goto err3;
        inet_ok:;
                hack_register_fcb(orb, port);
        }
 #endif
        return orb;
 
+err3:   forb_executor_destroy(&forb->default_executor);
 err2:  forb_free(forb);
 err:   return NULL;
 }
@@ -295,6 +300,29 @@ void forb_destroy(forb_orb orb)
        forb_t *forb = forb_object_to_forb(orb);
        forb_port_t *port;
        forb_regref_t *regref;
+       forb_object obj;
+
+       /* Stop ports to prevent remote requests from coming */
+       ul_list_for_each(forb_port, forb, port) {
+               forb_stop_port(port);
+       }
+
+       /* Wait for executors to finish all requests (and thus drop
+        * all references to peers). This is very inefficient for big
+        * number of objects, but we do not care */
+       gavl_cust_for_each(forb_objects_nolock, forb, obj) {
+               forb_executor_t *executor;
+               executor = forb_object_get_executor(obj);
+               if (executor)
+                       forb_executor_synchronize(executor);
+       }
+
+       /* Destroy ports - this should drop all remaining references
+        * to peers and result in closing of all remote
+        * connections. */
+       ul_list_for_each_cut(forb_port, forb, port) {
+               forb_destroy_port(port);
+       }
 
        pthread_cancel(forb->execution_thread.pthread_id);
        pthread_join(forb->execution_thread.pthread_id, NULL);
@@ -304,9 +332,6 @@ void forb_destroy(forb_orb orb)
                forb_unregister_reference(orb, regref->name.str);
        }
        
-       ul_list_for_each_cut(forb_port, forb, port) {
-               forb_destroy_port(port);
-       }
        forb_object_release(orb);
        forb_free(forb);
 }