@@ -1,30 +1,35 @@

-1.2.0:

- - log format change. add "[error]" for error log

- - latency improvement.

- - fix: terminate process when file close error.

-

-1.1.3:

+1.2.0: 2009/5/25

+ - Improved response.

+ - Added "[error]" to error logging messages.

+ - Added detailed status message in the event of member timeout.

+ - The msync's status option message now includes the command line of current process.

+ - Fixed abnormal behavior when multiple msyncs are invoked simultaneously.

+ - Fixed abnormal termination after file close failure.

+ - Fixed message output when the delete option is used.

+ - Fixed some minor flaws. 

+

+1.1.3: 2009/1/14

  - sort msync --members by hostname

  - fix: msync failed for FreeBSD and MacOSX

-1.1.2:

+1.1.2: 2008/12/24

  - fix: not send shutdown message 

-1.1.1:

+1.1.1: 2008/12/24

  - error message output in detail

-1.1.0:

+1.1.0: 2008/12/24

  - file transfer performance tuning

  - remote file delete function for makuosan

  - add --delete option for msync

  - add --sync   option for msync

-1.0.1:

+1.0.1: 2008/11/12

  - add spec file   (support/makuosan.spec)

  - add init script (support/makuosan.sysv)

  - support for FreeBSD & MacOSX

-1.0.0:

+1.0.0: 2008/11/6

  - stable release

  - change default value -f option. 1 to 3

  - fixed group missmatch

@@ -1,27 +1,32 @@

 1.2.0: 2009/5/25

+ - 応答速度が向上した

  - エラーログに"[error]"という文字列を追加

- - 転送中にメンバがタイムアウトしたときに状況を詳細に出力するようにした

- - msyncのstatusオプションで、実行中のコマンドも表示するようにした

+ - 転送中にメンバがタイムアウトしたときの状況を詳細に出力するようにした

+ - msync --statusで、現在実行中のコマンドも表示するようにした

  - msyncを複数同時実行したときに挙動がおかしくなる不具合を修正

  - ファイルクローズが失敗したときに異常終了する不具合を修正

  - msyncにdeleteオプションを指定したときの結果表示がおかしくなる不具合を修正

  - その他細かい不具合を修正

 1.1.3: 2009/1/14

- - FreeBSDから msync -r ができなかった不具合を修正

+ - FreeBSDで msync -r ができなかった不具合を修正

  - msync --membersの結果をソートするようにした

 1.1.2: 2008/12/24

  - 終了時にシャットダウンメッセージを飛ばしていなかった不具合を修正

 1.1.1: 2008/12/24

- - msyncコマンドに--deleteオプションを追加

- - 転送速度が向上（当社比1.5倍）

  - エラーを詳細に表示するようにした

+1.1.0: 2008/12/24

+ - msyncコマンドにdeleteオプションを追加

+ - msyncコマンドにsyncオプションを追加

+ - 転送速度が向上（当社比1.5倍）

+

 1.0.1: 2008/11/12

  - FreeBSDとMacOSXに対応

  - SPECファイルと起動スクリプトを同梱

 1.0.0: 2008/11/6

  - 安定版リリース

+

@@ -10,6 +10,7 @@ mfile *mfreeobj = NULL;

 mhost *members  = NULL;

 int loop_flag   = 1;

 struct timeval curtime;

+struct timeval lastpong;

 BF_KEY EncKey;

 int md5sum(int fd, unsigned char *digest)

@@ -31,7 +32,7 @@ int md5sum(int fd, unsigned char *digest)

 /*

  *  タイムアウト時間が経過しているかどうかを判断する

  *   - 現在時刻がtfからmsec[ms]経過していれば1を返す

- *   - それ以外は0を返す

+ *   - 経過していなければ0を返す

  */

 int mtimeout(struct timeval *tf, uint32_t msec)

 {

@@ -579,7 +580,7 @@ int ack_check(mfile *m, int state)

     if(!m->sendto){

       s = get_hoststate(t,m);

       if(!s){

-        lprintf(0,"%s: can't get state area host=%s fn=%s\n", 

+        lprintf(0,"[error] %s: can't get state area host=%s fn=%s\n", 

           __func__, t->hostname, m->fn);

       }else{

         if(*s == state){

@@ -590,7 +591,7 @@ int ack_check(mfile *m, int state)

       if(!memcmp(&(m->addr.sin_addr), &(t->ad), sizeof(m->addr.sin_addr))){

         s = get_hoststate(t,m);

         if(!s){

-          lprintf(0,"%s: can't get state area host=%s fn=%s\n", 

+          lprintf(0,"[error] %s: can't get state area host=%s fn=%s\n", 

             __func__, t->hostname, m->fn);

         }else{

           if(*s == state){

@@ -24,30 +24,16 @@ void recv_timeout(mfile *m)

   m->retrycnt = MAKUO_SEND_RETRYCNT;

 }

-struct timeval *pingpong(int n)

+void pingpong(int n)

 {

-  static struct timeval tv;

   mfile *m = mfins(MFSEND);

   mping *p = NULL;

   char buff[MAKUO_HOSTNAME_MAX + 1];

-  gettimeofday(&tv, NULL);

   if(!m){

-    lprintf(0, "%s: out of memmory\r\n", __func__);

-    return(&tv);

+    lprintf(0, "[error] %s: out of memmory\r\n", __func__);

+    return;

   }

-  switch(n){

-    case 0:

-      m->mdata.head.opcode = MAKUO_OP_PING;

-      break;

-    case 1:

-      m->mdata.head.opcode = MAKUO_OP_PING;

-      m->mdata.head.flags |= MAKUO_FLAG_ACK;

-      break;

-    case 2:

-      m->mdata.head.opcode = MAKUO_OP_EXIT;

-      break;

-  } 

   m->mdata.head.reqid  = getrid();

   m->mdata.head.seqno  = 0;

   m->mdata.head.szdata = 0;

@@ -66,22 +52,41 @@ struct timeval *pingpong(int n)

   m->mdata.p += p->versionlen;

   p->hostnamelen = htons(p->hostnamelen);

   p->versionlen  = htons(p->versionlen);

-  return(&tv);

+  gettimeofday(&lastpong, NULL);

+  switch(n){

+    case 0:

+      m->mdata.head.opcode = MAKUO_OP_PING;

+      break;

+    case 1:

+      m->mdata.head.opcode = MAKUO_OP_PING;

+      m->mdata.head.flags |= MAKUO_FLAG_ACK;

+      break;

+    case 2:

+      m->mdata.head.opcode = MAKUO_OP_EXIT;

+      msend(m);

+      break;

+  } 

 }

-int mcomm_accept(mcomm *c, fd_set *fds, int s)

+int mcomm_accept(mcomm *c, fd_set *fds)

 {

   int i;

-  if(s == -1)

+  int s = moption.lisocket;

+  if(s == -1){

     return(0);

-  if(!FD_ISSET(s,fds))

+  }

+  if(!FD_ISSET(s,fds)){

     return(0);

-  for(i=0;i<MAX_COMM;i++)

-    if(c[i].fd[0] == -1)

+  }

+  for(i=0;i<MAX_COMM;i++){

+    if(c[i].fd[0] == -1){

       break;

+    }

+  }

   if(i==MAX_COMM){

     close(accept(s, NULL, 0)); 

-    return(0);

+    lprintf(0, "[error] %s: can't accept reached in the maximum\n");

+    return(1);

   }

   c[i].addrlen = sizeof(c[i].addr);

   c[i].fd[0] = accept(s, (struct sockaddr *)(&c[i].addr), &(c[i].addrlen));

@@ -122,22 +127,25 @@ int mcomm_read(mcomm *c, fd_set *fds){

   return(0);

 }

-int mcomm_fdset(mcomm *c, fd_set *fds)

+int mcomm_fdset(mcomm *c, fd_set *rfds, fd_set *wfds)

 {

   int i;

   /*----- listen socket -----*/

   if(moption.lisocket != -1){

-    FD_SET(moption.lisocket, fds);

+    FD_SET(moption.lisocket, rfds);

   }

   /*----- connect socket -----*/

   for(i=0;i<MAX_COMM;i++){

     if(c[i].fd[0] != -1){

-      FD_SET(c[i].fd[0], fds);

+      FD_SET(c[i].fd[0], rfds);

+      if(c[i].working){

+        FD_SET(c[i].fd[0], wfds);

+      }

     }

     if(c[i].fd[1] != -1){

-      FD_SET(c[i].fd[1], fds);

+      FD_SET(c[i].fd[1], rfds);

     }else{

       if(c[i].cpid){

         if(waitpid(c[i].cpid, NULL, WNOHANG) == c[i].cpid){

@@ -180,9 +188,8 @@ int mfdirchk(mfile *d){

   return(1);

 }

-int ismsend(mfile *m, int flag)

+int is_send(mfile *m)

 {

-  int r;

   if(!m){

     return(0);

   }

@@ -190,108 +197,126 @@ int ismsend(mfile *m, int flag)

     return(0);

   }

   if(m->mdata.head.flags & MAKUO_FLAG_ACK){

-    if(flag){

-      msend(m);

-      return(0);

-    }

     return(1);

   }

-  if(!S_ISLNK(m->fs.st_mode) && S_ISDIR(m->fs.st_mode)){

-    if(!mfdirchk(m)){

-      return(0);

-    }

-  }

-  r = ack_check(m, MAKUO_RECVSTATE_NONE);

-  if(r == -1){

-    m->mdata.head.seqno  = 0;

-    m->mdata.head.nstate = MAKUO_SENDSTATE_BREAK;

+  switch(m->mdata.head.opcode){

+    case MAKUO_OP_SEND:

+    case MAKUO_OP_DEL:

+      if(!S_ISLNK(m->fs.st_mode) && S_ISDIR(m->fs.st_mode)){

+        if(!mfdirchk(m)){

+          return(0);

+        }

+      }

+      break;

   }

-  if(!r){

+  if(!ack_check(m, MAKUO_RECVSTATE_NONE)){

     m->sendwait = 0;

   }

   if(m->sendwait){

     if(!mtimeout(&(m->lastsend), MAKUO_SEND_TIMEOUT)){

-      return(1);

+      return(0);

     }

     if(!(m->retrycnt)){

       recv_timeout(m);

     }

   }

-  if(flag){

-    msend(m);

-  }

   return(1);

 }

-void mloop()

+void wfdset(int s, fd_set *fds)

 {

-  int para;

-  mfile *n;

   mfile *m;

-  fd_set rfds;

-  fd_set wfds;

-  struct timeval *lastpong;

+  for(m=mftop[MFSEND];m;m=m->next){

+    if(is_send(m)){

+      FD_SET(s, fds);

+      return;

+    }

+  }

+}

+

+int do_select(fd_set *rfds, fd_set *wfds)

+{

   struct timeval tv;

+  tv.tv_sec  = 1;

+  tv.tv_usec = 0;

+  if(select(1024, rfds, wfds, NULL, &tv) <= 0){

+    gettimeofday(&curtime, NULL);

+    moption.sendready = 0;

+    mrecv_gc();

+    return(-1);

+  }

+  gettimeofday(&curtime, NULL);

+  moption.sendready = FD_ISSET(moption.mcsocket, wfds);

+  return(0);

+}

-  /* multicast ping request */  

-  lastpong = pingpong(0);

+void do_pong()

+{

+  if(mtimeout(&lastpong, MAKUO_PONG_INTERVAL)){

+    pingpong(1);

+  }

+}

-  /* main loop */

-  while(loop_flag){

-    gettimeofday(&curtime, NULL);

-    if(mtimeout(lastpong, MAKUO_PONG_INTERVAL)){

-      lastpong = pingpong(1);

-    }

+void do_recv()

+{

+  while(mrecv());

+}

-    /* udp packet receive */

-    while(mrecv());

+void do_send()

+{

+  int  i=0;

+  mfile *m;

+  mfile *n;

-    /* udp packet send */

-    para = 0;

-    m = mftop[MFSEND];

-    while(m){

-      n = m->next;

-      para += ismsend(m, 1);

-      m = n;

-      if(para == moption.parallel){

-        break;

+  for(m=mftop[MFSEND];m;m=n){

+    if(i == moption.parallel){

+      return;

+    }

+    n = m->next;

+    if(m->mdata.head.flags & MAKUO_FLAG_ACK){

+      msend(m);

+      continue;

+    }

+    if(!is_send(m)){

+      if(m->sendwait){

+        i++;

       }

+      continue;

     }

+    msend(m);

+    i++;

+  }

+}

-    /* command check */

-    mcomm_check(moption.comm);

-

-    /* wait */

+void mloop()

+{

+  fd_set rfds;

+  fd_set wfds;

+  while(loop_flag){

     FD_ZERO(&rfds);

     FD_ZERO(&wfds);

-    FD_SET(moption.mcsocket,  &rfds);

-    mcomm_fdset(moption.comm, &rfds);

-    for(m=mftop[MFSEND];m;m=m->next){

-      if(ismsend(m, 0)){

-        FD_SET(moption.mcsocket, &wfds);

-        break;

-      }

-    }

-    tv.tv_sec  = 1;

-    tv.tv_usec = 0;

-    if(select(1024, &rfds, &wfds, NULL, &tv) == -1){

-      mrecv_gc();

+    FD_SET(moption.mcsocket, &rfds);

+    wfdset(moption.mcsocket, &wfds);

+    mcomm_fdset(moption.comm, &rfds, &wfds);

+    if(do_select(&rfds, &wfds)){

+      do_pong();

       continue;

     }

-    moption.sendready = FD_ISSET(moption.mcsocket,&wfds);

-    mcomm_accept(moption.comm, &rfds, moption.lisocket);  /* new console  */

-    mcomm_read(moption.comm, &rfds);                      /* command exec */

+    do_pong();

+    do_recv();

+    do_send();

+    mcomm_check(moption.comm);         /* exec check   */

+    mcomm_accept(moption.comm, &rfds); /* new console  */

+    mcomm_read(moption.comm, &rfds);   /* command exec */

   }

-

-  /* shutdown notify */

-  pingpong(2);

-  msend(mftop[MFSEND]);

 }

 int main(int argc, char *argv[])

 {

   minit(argc,argv);

+  pingpong(0);

   mloop();

+  pingpong(2);

   mexit();

   return(0);

 }

@@ -289,6 +289,7 @@ extern char *tzname[2];

 extern int daylight;

 extern char TZ[256];

 extern struct timeval curtime;

+extern struct timeval lastpong;

 extern BF_KEY EncKey;

 /*----- report -----*/

@@ -25,7 +25,7 @@ mfile *mexec_with_dsync(mcomm *c, char *fn, int dryrun, int recurs, mhost *t)

 {

   mfile *m = mfadd(MFSEND);

   if(!m){

-	  lprintf(0, "%s: out of memorry\n", __func__);

+	  lprintf(0, "[error] %s: out of memorry\n", __func__);

 	  cprintf(0, c, "error: out of memorry\n");

     return(m);

 	}

@@ -86,8 +86,7 @@ int mexec_scan_cmd(int fd, char *buff)

         if(errno == EINTR){

           continue;

         }

-        lprintf(0, "%s: commend write error (%s) %s", 

-          __func__, strerror(errno), buff);

+        lprintf(0, "[error] %s: commend write error (%s) %s", __func__, strerror(errno), buff);

         return(-1);

       }

       size -= r;

@@ -416,7 +415,7 @@ int mexec_send(mcomm *c, int n, int sync)

   /*----- send file -----*/

   m = mfadd(MFSEND);

   if(!m){

-	  lprintf(0, "%s: out of memorry\n", __func__);

+	  lprintf(0, "[error] %s: out of memorry\n", __func__);

 	  cprintf(0, c, "error: out of memorry\n");

     return(0);

 	}

@@ -1000,7 +999,7 @@ int mexec(mcomm *c, int n)

     if(r>0){

     }else{

       if(r == -1){

-        lprintf(0, "%s: read error n=%d\n", __func__, n);

+        lprintf(0, "[error] %s: read error n=%d\n", __func__, n);

       }

       mexec_close(c, n);

     }

@@ -1008,7 +1007,7 @@ int mexec(mcomm *c, int n)

   }

   if(n == 1){

-    for(m=mftop[0];m;m=m->next){

+    for(m=mftop[MFSEND];m;m=m->next){

       if(m->comm == c){

         count++;

         if(count == MAKUO_PARALLEL_MAX){

@@ -1019,7 +1018,7 @@ int mexec(mcomm *c, int n)

   }

   if(!size){

-    lprintf(0, "%s: buffer over fllow n=%d\n", __func__, n);

+    lprintf(0, "[error] %s: buffer over fllow n=%d\n", __func__, n);

     mexec_close(c, n);

     return(-1);

   }

@@ -1086,6 +1086,7 @@ static void mrecv_req_dsync_open(mfile *m, mdata *data, struct sockaddr_in *addr

     memcpy(m->fn, data->data, data->head.szdata);

   }

   m->fn[data->head.szdata] = 0;

+  mtimeget(&(m->lastrecv));

 }

 static void mrecv_req_dsync_data(mfile *m, mdata *data, struct sockaddr_in *addr)

@@ -1097,6 +1098,11 @@ static void mrecv_req_dsync_data(mfile *m, mdata *data, struct sockaddr_in *addr

   uint16_t len;

   msend(mkack(data, addr, MAKUO_RECVSTATE_OPEN));

+  if(!m){

+    return;

+  }

+

+  mtimeget(&(m->lastrecv));

   if(m->mdata.head.seqno >= data->head.seqno){

     return;

   }

@@ -1162,18 +1168,20 @@ static void mrecv_req_dsync_close(mfile *m, mdata *data, struct sockaddr_in *add

 {

   if(!m){

     msend(mkack(data, addr, MAKUO_RECVSTATE_CLOSE));

-    return;

-  }

-  if(m->link){

-    msend(mkack(data, addr, MAKUO_RECVSTATE_OPEN));

   }else{

-    msend(mkack(data, addr, MAKUO_RECVSTATE_CLOSE));

-    mrecv_mfdel(m); 

+    mtimeget(&(m->lastrecv));

+    if(m->link){

+      msend(mkack(data, addr, MAKUO_RECVSTATE_OPEN));

+    }else{

+      msend(mkack(data, addr, MAKUO_RECVSTATE_CLOSE));

+      mrecv_mfdel(m); 

+    }

   }

 }

 static void mrecv_req_dsync_break(mfile *m, mdata *data, struct sockaddr_in *addr)

 {

+  msend(mkack(data, addr, MAKUO_RECVSTATE_BREAK));

   if(m){

     if(m->link){

       m->link->mdata.head.nstate = MAKUO_SENDSTATE_BREAK;

@@ -1181,7 +1189,6 @@ static void mrecv_req_dsync_break(mfile *m, mdata *data, struct sockaddr_in *add

     }

     mrecv_mfdel(m);

   }

-  msend(mkack(data, addr, MAKUO_RECVSTATE_BREAK));

 }

 /*

@@ -807,11 +807,12 @@ static void msend_req_dsync_close(int s, mfile *m)

     msend_packet(s, &(m->mdata), &(m->addr));

     return;

   }

-  if(!ack_check(m, MAKUO_RECVSTATE_OPEN)){

-    msend_mfdel(m);

-  }else{

+  if(ack_check(m, MAKUO_RECVSTATE_OPEN)){

     m->sendwait  = 0;

     m->initstate = 1;

+  }else{

+    msend_packet(s, &(m->mdata), &(m->addr));

+    msend_mfdel(m);

   }

 }

@@ -144,7 +144,7 @@ int connect_socket(char *target)

     p = strtok(NULL,":");

     return(connect_socket_unix(p));

   }

-  return(-1);

+  return(connect_socket_unix(buff));

 }

 int writeline(int s, char *buff)

@@ -1,5 +1,5 @@

 Name:           makuosan

-Version:        1.1.4

+Version:        1.2.0

 Release:        1%{?dist}

 Summary:        Multicasts All-Kinds of Updating Operation for Servers on Administered Network

@@ -83,7 +83,8 @@ rm -rf %{buildroot}

 %changelog

-* Sat May 16 2009 Masanobu Yasui <yasui0906@gmail.com> - 1.1.4

+* Mon May 25 2009 Masanobu Yasui <yasui0906@gmail.com> - 1.2.0

 * Thu Nov  6 2008 Naoya Nakazawa <naoya.n@gmail.com> - 1.0.0

 - Initial version

+

@@ -188,12 +188,13 @@ void mprintf(int l, const char *func, mfile *m)

 {

   if(!m)

     return;

-  lprintf(l, "%s: rc=%d rid=%d init=%d wait=%d %s %s %s %s\n",

+  lprintf(l, "%s: rc=%d id=%d init=%d wait=%d flag=%d %s %s %s %s\n",

     func, 

     m->retrycnt,

     m->mdata.head.reqid, 

     m->initstate, 

     m->sendwait, 

+    m->mdata.head.flags,

     inet_ntoa(m->addr.sin_addr), 

     stropcode(&(m->mdata)),

     strmstate(&(m->mdata)),