Custom config memberlist (kleveross#68)

* fix: mismatch between time.Duration and C.Int

* feat: improve join process in memberlist

* feat: expose joined status

Author: zw0610

ftlib/consensus/gossip/impl.py

@@ -46,10 +46,11 @@ def __init__(
         if res != 0:
             raise RuntimeError("failed to initialize memberlist")
 
+        self.joined = False
         if known_addr_list is not None and known_addr_list != []:
-            joined = self._join(known_addr_list=known_addr_list)
-            if not joined:
-                raise RuntimeError("failed to join the group")
+            self.joined = self._join(known_addr_list=known_addr_list)
+            if not self.joined:
+                logging.warning("failed to join the group")
 
         # we need to take a sleep here because the group may add further nodes
         # after this node succeeded joining the group
@@ -79,7 +80,12 @@ def get_memberlist(self):
     def passive_or_active(self):
         return ConsensusMode.ACTIVE
 
-    def _join(self, known_addr_list, codec="utf-8"):
+    def manual_join(self, known_addr_list, wait_time=15):
+        # return bool value: self._joined
+        self.joined = self._join(known_addr_list, wait_time=wait_time)
+        return self.joined
+
+    def _join(self, known_addr_list, codec="utf-8", wait_time=15):
         assert type(known_addr_list) == list
 
         addr_list_len = len(known_addr_list)
@@ -98,7 +104,7 @@ def _join(self, known_addr_list, codec="utf-8"):
         # TODO: waiting for 15 sec is not an optimal choice
         #  it will better to set 15 sec as max_timeout, and
         #  the stop condition is
-        time.sleep(15)
+        time.sleep(wait_time)
         res = self._lib.join(t)
 
         return res > 0

ftlib/consensus/gossip/memberlist/main.go

@@ -34,17 +34,21 @@ func init_memberlist(
     logFileName := C.GoString(cLogFileName)
     customBindAddr := C.GoString(cBindAddr)
     customAdvertiseAddr := C.GoString(cAdvertiseAddr)
+    customTCPTimeout := time.Duration(int64(cTCPTimeout))
 
     if customBindAddr != "" {
+        fmt.Printf("Setting Bind Address as %s\n", customBindAddr)
         config.BindAddr = customBindAddr
     }
 
     if customAdvertiseAddr != "" {
+        fmt.Printf("Setting Advertise Address as %s\n", customAdvertiseAddr)
         config.AdvertiseAddr = customAdvertiseAddr
     }
 
-    if cTCPTimeout > 0 {
-        config.TCPTimeout = cTCPTimeout * time.Second
+    if int64(cTCPTimeout) > 0 {
+        fmt.Printf("Setting TCPTimeout as %d\n", int64(cTCPTimeout))
+        config.TCPTimeout = customTCPTimeout * time.Second
     }
 
     if logFileName != "" {
@@ -74,7 +78,7 @@ func join(ns []*C.char) C.int {
     }
     n, err := list.Join(addrList)
     if err != nil {
-        panic("Failed to join cluster: " + err.Error())
+        fmt.Print("Failed to join cluster: " + err.Error())
         return 0
     }
     return C.int(n)

ftlib/impl.py

@@ -133,6 +133,12 @@ def set_initialized(self, is_initialized):
             self.commlib.abort_communicator()
             self.consensus.average_failure()
 
+    def manual_join(self, *args, **kwargs):
+        return self.consensus.manual_join(*args, **kwargs)
+
+    def consensus_joined(self):
+        return self.consensus.joined
+
     # TODO: execute still under development
     def execute(self, func, *args, **kwargs):
         # Args: