]> rtime.felk.cvut.cz Git - frescor/forb.git/blobdiff - src/forb.c
forb: Split forb_port_destroy() to stop and destroy phases
[frescor/forb.git] / src / forb.c
index dc55cbcfa431f3bdf37235246ecf3df12e18927b..309c6602e8f8b5a84b469be748058385e5b76e5e 100644 (file)
@@ -74,6 +74,7 @@
 #include <ul_log.h>
 #include <ul_logreg.h>
 #include <unistd.h>
+#include "discovery.h"
 #ifdef CONFIG_FORB_PROTO_UNIX
 #include <forb/proto_unix.h>
 #endif
@@ -84,6 +85,7 @@
 #include <fcb.h>
 #include <fcb_contact_info.h>
 #endif
+#include <fcntl.h>
 
 #ifdef DEBUG
 #define UL_LOGL_DEF UL_LOGL_DEB
 #include "log_domains.inc"
 
 extern UL_LOG_CUST(ulogd_forb);
+static int init_ul_log(void);
 
 UL_LOGREG_DOMAINS_INIT_FUNCTION(forb_logreg_domains, forb_logreg_domains_array);
 
+#if 0
 static void
 destroy_forb_on_exit(int exitcode, void *arg)
 {
        forb_orb orb = arg;
        forb_destroy(orb);
 }
+#endif
 
 static void
 forb_is_alive(forb_orb _obj, CORBA_Environment *ev)
@@ -145,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;
 }
 
@@ -180,6 +184,7 @@ forb_init(int *argc, char **argv[], const struct forb_init_attr *attr)
        memset(forb, 0, sizeof(*forb));
 
        /* Initialize ULUT logging facility */
+       init_ul_log();
        forb_logreg_domains();
 
        if (attr) {
@@ -188,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 {
@@ -203,12 +210,8 @@ forb_init(int *argc, char **argv[], const struct forb_init_attr *attr)
        
        if (fosa_mutex_init(&forb->request_mutex, 0) != 0) goto err2;
        forb_request_nolock_init_root_field(forb);
-       
-       if (fosa_mutex_init(&forb->peer_mutex, 0) != 0) goto err2;
-       forb_peer_nolock_init_root_field(forb);
 
-       if (fosa_mutex_init(&forb->objkey_mutex, 0) != 0) goto err2;
-       forb_objects_nolock_init_root_field(forb);
+       if (forb_discovery_init(forb) != 0) goto err2;
 
        if (fosa_mutex_init(&forb->regref_mutex, 0) != 0) goto err2;
        forb_regref_nolock_init_root_field(forb);
@@ -219,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 */
@@ -230,12 +235,19 @@ 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;
-       
-       on_exit(destroy_forb_on_exit, orb);
+               goto err3;
+
+       /* FIXME: I do not know how to deregister the exit handler if
+        * forb_destroy() is called manually. */
+       /* on_exit(destroy_forb_on_exit, orb); */
 
 #ifdef CONFIG_FORB_PROTO_UNIX
        {
@@ -250,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
@@ -271,25 +283,46 @@ 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;
 }
 
-/* FIXME: forb_destroy is now called automatically on exit. Therefore
- * forb_destroy() should be either static or should deregister on_exit
- * handler (how???).  */
 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);
@@ -299,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);
 }
@@ -485,7 +515,8 @@ forb_get_req_source(const forb_object obj, forb_server_id *req_source)
  * @param num_elements
  * @param elem_size
  *
- * @return CORBA_TRUE if the allocation was sucessfull, CORBA_FALSE if wasn't.
+ * @return CORBA_TRUE if the allocation was sucessfull, CORBA_FALSE if
+ * it wasn't.
  */
 CORBA_boolean
 __forb_sequence_alloc_buf_internal(void *seq, size_t seq_size,
@@ -499,3 +530,103 @@ __forb_sequence_alloc_buf_internal(void *seq, size_t seq_size,
        *maximum_ptr = *buf_pptr ? num_elements : 0;
        return (*buf_pptr != NULL) || (num_elements == 0);
 }
+
+static FILE *forb_ul_log_file;
+static char progname[64] = "";
+
+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)
+                       fprintf(forb_ul_log_file,"%s: ",domain->name);
+       }
+       vfprintf(forb_ul_log_file, format, ap);
+       fflush(forb_ul_log_file);
+}
+
+static int init_ul_log(void)
+{
+       char *s;
+       char *log_fname;
+       int fd;
+       int flg = 0;
+       char path[128];
+
+/*     if(ul_log_output != NULL) */
+/*             return 0; */
+
+       fd = open("/proc/self/cmdline", O_RDONLY);
+       if (fd >= 0) {
+               int ret;
+               ret = read(fd, path, sizeof(path)-1);
+               if (ret > 0) {
+                       path[ret]=0;
+                       s = strrchr(path, '/');
+                       if (s) s++;
+                       else s = path;
+                       strncpy(progname, s, sizeof(progname)-1);
+               }
+               close(fd);
+       }
+
+       if((log_fname=getenv("UL_LOG_FILENAME"))!=NULL){
+               forb_ul_log_file=fopen(log_fname,"a");
+       }
+       if(forb_ul_log_file==NULL)
+               forb_ul_log_file=stderr;
+       
+       if((s=getenv("UL_DEBUG_FLG")) != NULL){
+               flg=atoi(s);
+       }
+
+       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);
+}