NSYN: adjust to new semantics
[citadel.git] / citadel / modules / network / serv_network.c
index 599b90d1d8d820070b514c442973b03e71dcc0a1..5ab8892a0afa3ff5ae04b73952888800e2974809 100644 (file)
@@ -2,7 +2,7 @@
  * This module handles shared rooms, inter-Citadel mail, and outbound
  * mailing list processing.
  *
- * Copyright (c) 2000-2011 by the citadel.org team
+ * Copyright (c) 2000-2012 by the citadel.org team
  *
  *  This program is open source software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -112,7 +112,7 @@ int GetNetworkedRoomNumbers(const char *DirName, HashList *DirList)
        struct dirent *d;
        struct dirent *filedir_entry;
        long RoomNR;
-       long Count;
+       long Count = 0;
                
        filedir = opendir (DirName);
        if (filedir == NULL) {
@@ -204,7 +204,8 @@ int network_usetable(struct CtdlMessage *msg) {
  * ignoring anything we know about which messages have already undergone
  * network processing.  This can be used to bring a new node into sync.
  */
-int network_sync_to(char *target_node) {
+int network_sync_to(char *target_node, long len)
+{
        SpoolControl sc;
        int num_spooled = 0;
        int found_node = 0;
@@ -223,33 +224,41 @@ int network_sync_to(char *target_node) {
                end_critical_section(S_NETCONFIGS);
                return(-1);
        }
-       while (fgets(buf, sizeof buf, fp) != NULL) {
+       while (fgets(buf, sizeof buf, fp) != NULL)
+       {
                buf[strlen(buf)-1] = 0;
+
                extract_token(sc_type, buf, 0, '|', sizeof sc_type);
+               if (strcasecmp(sc_type, "ignet_push_share"))
+                       continue;
+
                extract_token(sc_node, buf, 1, '|', sizeof sc_node);
+               if (!strcasecmp(sc_node, target_node))
+                       continue;
+
                extract_token(sc_room, buf, 2, '|', sizeof sc_room);
-               if ( (!strcasecmp(sc_type, "ignet_push_share"))
-                  && (!strcasecmp(sc_node, target_node)) ) {
-                       found_node = 1;
+               found_node = 1;
                        
-                       /* Concise syntax because we don't need a full linked-list */
-                       memset(&sc, 0, sizeof(SpoolControl));
-                       sc.ignet_push_shares = (maplist *)
-                               malloc(sizeof(maplist));
-                       sc.ignet_push_shares->next = NULL;
-                       safestrncpy(sc.ignet_push_shares->remote_nodename,
-                               sc_node,
-                               sizeof sc.ignet_push_shares->remote_nodename);
-                       safestrncpy(sc.ignet_push_shares->remote_roomname,
-                               sc_room,
-                               sizeof sc.ignet_push_shares->remote_roomname);
-               }
+               /* Concise syntax because we don't need a full linked-list */
+               memset(&sc, 0, sizeof(SpoolControl));
+               sc.ignet_push_shares = (maplist *)
+                       malloc(sizeof(maplist));
+               sc.ignet_push_shares->next = NULL;
+               safestrncpy(sc.ignet_push_shares->remote_nodename,
+                           sc_node,
+                           sizeof sc.ignet_push_shares->remote_nodename);
+               safestrncpy(sc.ignet_push_shares->remote_roomname,
+                           sc_room,
+                           sizeof sc.ignet_push_shares->remote_roomname);
        }
        fclose(fp);
        end_critical_section(S_NETCONFIGS);
 
        if (!found_node) return(-1);
 
+       sc.working_ignetcfg = load_ignetcfg();
+       sc.the_netmap = read_network_map();
+
        /* Send ALL messages */
        num_spooled = CtdlForEachMessage(MSGS_ALL, 0L, NULL, NULL, NULL,
                network_spool_msg, &sc);
@@ -257,6 +266,9 @@ int network_sync_to(char *target_node) {
        /* Concise cleanup because we know there's only one node in the sc */
        free(sc.ignet_push_shares);
 
+       DeleteHash(&sc.working_ignetcfg);
+       DeleteHash(&sc.the_netmap);
+
        syslog(LOG_NOTICE, "Synchronized %d messages to <%s>\n",
                num_spooled, target_node);
        return(num_spooled);
@@ -268,12 +280,13 @@ int network_sync_to(char *target_node) {
  */
 void cmd_nsyn(char *argbuf) {
        int num_spooled;
+       long len;
        char target_node[256];
 
        if (CtdlAccessCheck(ac_aide)) return;
 
-       extract_token(target_node, argbuf, 0, '|', sizeof target_node);
-       num_spooled = network_sync_to(target_node);
+       len = extract_token(target_node, argbuf, 0, '|', sizeof target_node);
+       num_spooled = network_sync_to(target_node, len);
        if (num_spooled >= 0) {
                cprintf("%d Spooled %d messages.\n", CIT_OK, num_spooled);
        }
@@ -306,7 +319,7 @@ void network_queue_interesting_rooms(struct ctdlroom *qrbuf, void *data) {
                ptr->namelen = ROOMNAMELEN - 1;
 
        memcpy (ptr->name, qrbuf->QRname, ptr->namelen);
-       ptr->name[ptr->namelen] = 0;
+       ptr->name[ptr->namelen] = '\0';
        ptr->QRNum = qrbuf->QRnumber;
 
        for (i = 0; i < ptr->namelen; i++)
@@ -314,6 +327,7 @@ void network_queue_interesting_rooms(struct ctdlroom *qrbuf, void *data) {
                ptr->lcname[i] = tolower(ptr->name[i]);
        }
 
+       ptr->lcname[ptr->namelen] = '\0';
        ptr->key = hashlittle(ptr->lcname, ptr->namelen, 9872345);
        ptr->next = RP->rplist;
        RP->rplist = ptr;
@@ -326,6 +340,8 @@ void network_queue_room(struct ctdlroom *qrbuf, void *data) {
        int i;
        struct RoomProcList *ptr;
 
+       if (qrbuf->QRdefaultview == VIEW_QUEUE)
+               return;
        ptr = (struct RoomProcList *) malloc(sizeof (struct RoomProcList));
        if (ptr == NULL) return;
 
@@ -334,15 +350,16 @@ void network_queue_room(struct ctdlroom *qrbuf, void *data) {
                ptr->namelen = ROOMNAMELEN - 1;
 
        memcpy (ptr->name, qrbuf->QRname, ptr->namelen);
-       ptr->name[ptr->namelen] = 0;
+       ptr->name[ptr->namelen] = '\0';
        ptr->QRNum = qrbuf->QRnumber;
 
        for (i = 0; i < ptr->namelen; i++)
        {
                ptr->lcname[i] = tolower(ptr->name[i]);
        }
-
+       ptr->lcname[ptr->namelen] = '\0';
        ptr->key = hashlittle(ptr->lcname, ptr->namelen, 9872345);
+
        begin_critical_section(S_RPLIST);
        ptr->next = rplist;
        rplist = ptr;
@@ -491,8 +508,8 @@ void network_do_queue(void) {
        static int doing_queue = 0;
        static time_t last_run = 0L;
        int full_processing = 1;
-       char *working_ignetcfg;
-       NetMap *the_netmap = NULL;
+       HashList *working_ignetcfg;
+       HashList *the_netmap = NULL;
        int netmap_changed = 0;
        roomlists RL;
 
@@ -524,7 +541,8 @@ void network_do_queue(void) {
        end_critical_section(S_RPLIST);
 
        RL.RoomsInterestedIn = NewHash(1, lFlathash);
-       if (!GetNetworkedRoomNumbers(ctdl_netcfg_dir, RL.RoomsInterestedIn))
+       if (full_processing &&
+           (GetNetworkedRoomNumbers(ctdl_netcfg_dir, RL.RoomsInterestedIn)==0))
        {
                doing_queue = 0;
                DeleteHash(&RL.RoomsInterestedIn);
@@ -532,13 +550,15 @@ void network_do_queue(void) {
                        return;
        }
        /* Load the IGnet Configuration into memory */
-       working_ignetcfg = load_working_ignetcfg();
+       working_ignetcfg = load_ignetcfg();
 
        /*
         * Load the network map and filter list into memory.
         */
-       the_netmap = read_network_map();
-       load_network_filter_list();
+       if (!server_shutting_down)
+               the_netmap = read_network_map();
+       if (!server_shutting_down)
+               load_network_filter_list();
 
        /* 
         * Go ahead and run the queue
@@ -548,7 +568,7 @@ void network_do_queue(void) {
                CtdlForEachRoom(network_queue_interesting_rooms, &RL);
        }
 
-       if (RL.rplist != NULL) {
+       if ((RL.rplist != NULL) && (!server_shutting_down)) {
                RoomProcList *ptr, *cmp;
                ptr = RL.rplist;
                syslog(LOG_DEBUG, "network: running outbound queue\n");
@@ -579,18 +599,27 @@ void network_do_queue(void) {
        /* If there is anything in the inbound queue, process it */
        if (!server_shutting_down) {
                network_do_spoolin(working_ignetcfg, 
-                                  &the_netmap,
+                                  the_netmap,
                                   &netmap_changed);
        }
 
-       /* Save the network map back to disk */
-       write_network_map(the_netmap, netmap_changed);
-
        /* Free the filter list in memory */
        free_netfilter_list();
 
+       /* Save the network map back to disk */
+       if (netmap_changed) {
+               StrBuf *MapStr = SerializeNetworkMap(the_netmap);
+               CtdlPutSysConfig(IGNETMAP, SmashStrBuf(&MapStr));
+       }
+
+       /* combine singe message files into one spool entry per remote node. */
        network_consolidate_spoolout(working_ignetcfg, the_netmap);
-       free(working_ignetcfg);
+
+       /* shut down. */
+
+       DeleteHash(&the_netmap);
+
+       DeleteHash(&working_ignetcfg);
 
        syslog(LOG_DEBUG, "network: queue run completed\n");
 
@@ -611,14 +640,102 @@ int network_room_handler (struct ctdlroom *room)
        return 0;
 }
 
+int NTTDebugEnabled = 0;
+
+/*
+ * network_talking_to()  --  concurrency checker
+ */
+static HashList *nttlist = NULL;
+int network_talking_to(const char *nodename, long len, int operation) {
+
+       int retval = 0;
+       HashPos *Pos = NULL;
+       void *vdata;
+
+       begin_critical_section(S_NTTLIST);
+
+       switch(operation) {
+
+               case NTT_ADD:
+                       if (nttlist == NULL) 
+                               nttlist = NewHash(1, NULL);
+                       Put(nttlist, nodename, len, NewStrBufPlain(nodename, len), HFreeStrBuf);
+                       if (NTTDebugEnabled) syslog(LOG_DEBUG, "nttlist: added <%s>\n", nodename);
+                       break;
+               case NTT_REMOVE:
+                       if ((nttlist == NULL) ||
+                           (GetCount(nttlist) == 0))
+                               break;
+                       Pos = GetNewHashPos(nttlist, 1);
+                       if (GetHashPosFromKey (nttlist, nodename, len, Pos))
+                               DeleteEntryFromHash(nttlist, Pos);
+                       DeleteHashPos(&Pos);
+                       if (NTTDebugEnabled) syslog(LOG_DEBUG, "nttlist: removed <%s>\n", nodename);
+
+                       break;
+
+               case NTT_CHECK:
+                       if ((nttlist == NULL) ||
+                           (GetCount(nttlist) == 0))
+                               break;
+                       if (GetHash(nttlist, nodename, len, &vdata))
+                               retval ++;
+                       if (NTTDebugEnabled) syslog(LOG_DEBUG, "nttlist: have [%d] <%s>\n", retval, nodename);
+                       break;
+       }
+
+       end_critical_section(S_NTTLIST);
+       return(retval);
+}
+
+void cleanup_nttlist(void)
+{
+        begin_critical_section(S_NTTLIST);
+       DeleteHash(&nttlist);
+        end_critical_section(S_NTTLIST);
+}
+
+
+
+void network_logout_hook(void)
+{
+       CitContext *CCC = MyContext();
+
+       /*
+        * If we were talking to a network node, we're not anymore...
+        */
+       if (!IsEmptyStr(CCC->net_node)) {
+               network_talking_to(CCC->net_node, strlen(CCC->net_node), NTT_REMOVE);
+               CCC->net_node[0] = '\0';
+       }
+}
+void network_cleanup_function(void)
+{
+       struct CitContext *CCC = CC;
+
+       if (!IsEmptyStr(CCC->net_node)) {
+               network_talking_to(CCC->net_node, strlen(CCC->net_node), NTT_REMOVE);
+               CCC->net_node[0] = '\0';
+       }
+}
+
 
 /*
  * Module entry point
  */
+void SetNTTDebugEnabled(const int n)
+{
+       NTTDebugEnabled = n;
+}
+
 CTDL_MODULE_INIT(network)
 {
        if (!threading)
        {
+               CtdlRegisterDebugFlagHook(HKEY("networktalkingto"), SetNTTDebugEnabled, &NTTDebugEnabled);
+               CtdlRegisterCleanupHook(cleanup_nttlist);
+               CtdlRegisterSessionHook(network_cleanup_function, EVT_STOP);
+                CtdlRegisterSessionHook(network_logout_hook, EVT_LOGOUT);
                CtdlRegisterProtoHook(cmd_nsyn, "NSYN", "Synchronize room to node");
                CtdlRegisterRoomHook(network_room_handler);
                CtdlRegisterCleanupHook(destroy_network_queue_room_locked);