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 dee15872072afbba395de57e31d456172ace6ab3..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;
 }
 
@@ -194,6 +193,8 @@ forb_init(int *argc, char **argv[], const struct forb_init_attr *attr)
                memset(&forb->attr, 0, sizeof(forb->attr));
        }
 
+       sem_init(&forb->server_ready, 0, 0);
+
        if (forb_server_id_empty(&forb->attr.fixed_server_id)) {
                forb_server_id_init(&forb->server_id);
        } else {
@@ -221,6 +222,8 @@ forb_init(int *argc, char **argv[], const struct forb_init_attr *attr)
                return NULL;
        }
 
+       forb_executor_prepare();        
+
        orb = forb_forb_orb_new(NULL, &forb_implementation, forb);
        if (!orb) goto err2;
        /* Server ID must be assigned manualy */
@@ -232,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. */
@@ -254,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
@@ -275,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;
 }
@@ -291,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);
@@ -300,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);
 }
@@ -509,8 +538,11 @@ void
 forb_ul_log_fnc(ul_log_domain_t *domain, int level,
                const char *format, va_list ap)
 {
+       struct timespec now;
        if(!(level&UL_LOGL_CONT)) {
                level&=UL_LOGL_MASK;
+               clock_gettime(CLOCK_MONOTONIC, &now);
+               fprintf(forb_ul_log_file,"%ld.%6ld: ", now.tv_sec, now.tv_nsec/1000);
                if (progname[0])
                        fprintf(forb_ul_log_file,"%s: ", progname);
                if(domain && domain->name)
@@ -557,5 +589,44 @@ static int init_ul_log(void)
 
        ul_log_redir(forb_ul_log_fnc, flg);
 
+       if((s = getenv("UL_LOG_LEVELS")) != NULL)
+               ul_log_domain_arg2levels(s);
+
        return 0;
 }
+
+/** 
+ * Wait for the server to be ready. Internal function intended forb
+ * forbrun.
+ * 
+ * @param orb ORB.
+ * 
+ * @return Zero on success; on error -1 is returned, and errno is set
+ * to indicate the error.
+ */
+int forb_wait_for_server_ready(forb_orb orb)
+{
+       forb_t *forb = forb_object_to_forb(orb);
+       return sem_wait(&forb->server_ready);
+}
+
+/** 
+ * Signal the the FORB core that the server is ready for accepting
+ * requests.
+ *
+ * This function should be called at the initialization of server
+ * implementation at the time when all objects are registered with
+ * executors. All other servers in the same address space are
+ * initialized after this function is called which allows the other
+ * servers to use the services provided by the calling server.
+ * 
+ * @param orb ORB object.
+ * 
+ * @return Zero on success; on error -1 is returned, and errno is set
+ * to indicate the error.
+ */
+int forb_signal_server_ready(forb_orb orb)
+{
+       forb_t *forb = forb_object_to_forb(orb);
+       return sem_post(&forb->server_ready);
+}