@@ -206,12 +206,18 @@ void fdprintf(int s, char *fmt, ...)

 void lprintf(int l, char *fmt, ...)

 {

   va_list arg;

+  struct timeval tv;

   char msg[2048];

   if(moption.loglevel >= l){

     va_start(arg, fmt);

     vsprintf(msg, fmt, arg);

     va_end(arg);

+#ifdef MAKUO_DEBUG

+    gettimeofday(&tv, NULL);

+    fprintf(stderr, "%02d.%06d %s", tv.tv_sec % 60, tv.tv_usec, msg);

+#else

     fprintf(stderr, "%s", msg);

+#endif

     syslog(LOG_ERR, "%s: %s", moption.user_name, msg);

   }

 }

@@ -233,7 +239,7 @@ void cprintf(int l, mcomm *c, char *fmt, ...)

   }

 }

-void mprintf(char *func, mfile *m)

+void mprintf(const char *func, mfile *m)

 {

   char *st;

   char *op;

@@ -131,17 +131,10 @@ int mcomm_accept(mcomm *c, fd_set *fds, int s)

   return(0);

 }

-int mcomm_read(mcomm *c, fd_set *fds){

+void mcomm_check(mcomm *c){

   int i, j;

   mfile *m;

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

-    for(j=0;j<2;j++){

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

-        if(FD_ISSET(c[i].fd[j], fds) || c[i].check[j]){

-          mexec(&c[i], j);

-        }

-      }

-    }

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

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

         if(m->comm == &c[i]){

@@ -155,6 +148,20 @@ int mcomm_read(mcomm *c, fd_set *fds){

       }

     }

   }

+}

+

+int mcomm_read(mcomm *c, fd_set *fds){

+  int i, j;

+  mfile *m;

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

+    for(j=0;j<2;j++){

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

+        if(FD_ISSET(c[i].fd[j], fds) || c[i].check[j]){

+          mexec(&c[i], j);

+        }

+      }

+    }

+  }

   return(0);

 }

@@ -185,8 +192,8 @@ int mcomm_fdset(mcomm *c, fd_set *fds)

 }

 int mfdirchk(mfile *d){

-  int len = strlen(d->fn);

   mfile *m;

+  int len = strlen(d->fn);

   if(!len){

     return(1);

   }

@@ -243,6 +250,9 @@ int ismsend(int s, mfile *m)

 /***** main loop *****/

 int mloop()

 {

+  int para;

+  mfile *n;

+  mfile *m;

   fd_set rfds;

   fd_set wfds;

   struct timeval *lastpong;

@@ -251,46 +261,41 @@ int mloop()

   gettimeofday(&curtime,NULL);

   lastpong = pingpong(0);

   while(loop_flag){

-    tv.tv_sec  = 1;

-    tv.tv_usec = 0;

-    FD_ZERO(&rfds);

-    FD_ZERO(&wfds);

-

     gettimeofday(&curtime,NULL);

     if(mtimeout(lastpong, MAKUO_PONG_INTERVAL))

       lastpong = pingpong(1);

-

-    mcomm_fdset(moption.comm, &rfds);

+    m = mftop[0];

+    while(mrecv(moption.mcsocket)){

+      if(m != mftop[0]){

+        break;

+      }

+    }

+    para = 0;

+    n = NULL;

+    m = mftop[0];

+    while(m){

+      n = m->next;

+      para += ismsend(moption.mcsocket, m);

+      m = n;

+      if(para == moption.parallel){

+        break;

+      }

+    }

+    mcomm_check(moption.comm);

+    FD_ZERO(&rfds);

+    FD_ZERO(&wfds);

     FD_SET(moption.mcsocket,  &rfds);

+    mcomm_fdset(moption.comm, &rfds);

     if(mftop[0]){

-      tv.tv_sec  = 0;

-      tv.tv_usec = 10000;

       FD_SET(moption.mcsocket, &wfds);

     }

-

-    if(select(1024, &rfds, &wfds, NULL, &tv) < 0)

+    tv.tv_sec  = 1;

+    tv.tv_usec = 0;

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

       continue;

-

-    gettimeofday(&curtime,NULL);

-    if(FD_ISSET(moption.mcsocket,&wfds)){

-      int para = 0;

-      mfile *n = NULL;

-      mfile *m = mftop[0];

-      while(m){

-        n = m->next;

-        para += ismsend(moption.mcsocket, m);

-        m = n;

-        if(para == moption.parallel){

-          break;

-        }

-      }

-    }

-    if(FD_ISSET(moption.mcsocket,&rfds))

-      mrecv(moption.mcsocket);

-

-    mrecv_gc();

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

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

+    mrecv_gc();

   }

   return(0);

 }

@@ -8,6 +8,7 @@

 #define PROTOCOL_VERSION 4

 #define _GNU_SOURCE

 #define _FILE_OFFSET_BITS 64

+#define MAKUO_DEBUG

 #include <stdio.h>

 #include <unistd.h>

 #include <stdlib.h>

@@ -55,8 +56,8 @@

 #define MAKUO_MCAST_PORT  5000

 /*----- timeout -----*/

-#define MAKUO_SEND_TIMEOUT  500    /* 再送間隔(ms)                                 */

-#define MAKUO_SEND_RETRYCNT 120    /* 再送回数                                     */

+#define MAKUO_SEND_TIMEOUT  250    /* 再送間隔(ms)                                 */

+#define MAKUO_SEND_RETRYCNT 240    /* 再送回数                                     */

 #define MAKUO_PONG_TIMEOUT  180000 /* メンバから除外するまでの時間(ms)             */

 #define MAKUO_PONG_INTERVAL 45000  /* PING送信間隔(ms)                             */

 #define MAKUO_RECV_GCWAIT   180000 /* 消し損ねたオブジェクトを開放する待ち時間(ms) */

@@ -222,8 +223,8 @@ typedef struct mfile

 typedef struct

 {

-  uint8_t state[MAKUO_PARALLEL_MAX];

-  mfile *mflist[MAKUO_PARALLEL_MAX];

+  uint8_t state[MAKUO_PARALLEL_MAX * MAX_COMM];

+  mfile *mflist[MAKUO_PARALLEL_MAX * MAX_COMM];

   char hostname[MAKUO_HOSTNAME_MAX];

   char version[32];

   struct in_addr ad;

@@ -278,7 +279,7 @@ extern BF_KEY EncKey;

 char    *SSTATE(uint8_t n);

 char    *RSTATE(uint8_t n);

 char    *OPCODE(uint8_t n);

-void     mprintf(char *func, mfile *m);

+void     mprintf(const char *func, mfile *m);

 void     lprintf(int l, char *fmt, ...);

 void     cprintf(int l, mcomm *c, char *fmt, ...);

 void     fdprintf(int s, char *fmt, ...);

@@ -293,7 +294,7 @@ mfile   *mkack(mdata *data, struct sockaddr_in *addr, uint8_t state);

 mhost   *member_get(struct in_addr *addr);

 mhost   *member_add(struct in_addr *addr, mdata *recvdata);

 void     member_del(mhost *h);

-void     mrecv(int s);

+int      mrecv(int s);

 void     msend(int s, mfile *m);

 void     set_filestat(char *path, uid_t uid, gid_t gid, mode_t mode);

 int      set_guid(int uid, int gid, gid_t *gids);

@@ -257,8 +257,9 @@ int mexec_send(mcomm *c, int n, int sync)

   char *argv[9];

   char *fn = NULL;

   mfile *m = NULL;

-  mhost *h = NULL;

-  int recursive = 0;

+  mhost *t = NULL;

+  int dryrun = 0;

+  int recurs = 0;

   int mode = MAKUO_MEXEC_SEND;

   if(moption.dontsend){

@@ -272,16 +273,17 @@ int mexec_send(mcomm *c, int n, int sync)

   while((i=getopt(c->argc[n], argv, "t:nr")) != -1){

     switch(i){

       case 'n':

+        dryrun = 1;

         mode = MAKUO_MEXEC_DRY;

         break;

       case 'r':

-        recursive = 1;

+        recurs = 1;

         break;

       case 't':

-        for(h=members;h;h=h->next)

-          if(!strcmp(h->hostname, optarg))

+        for(t=members;t;t=t->next)

+          if(!strcmp(t->hostname, optarg))

             break;

-        if(!h){

+        if(!t){

           cprintf(0, c, "%s is not contained in members\r\n", optarg);

           return(0);

         }

@@ -296,12 +298,12 @@ int mexec_send(mcomm *c, int n, int sync)

     fn = c->parse[n][optind++];

   /*----- directory scan -----*/

-  if(recursive){

+  if(recurs){

     if(c->cpid){

       cprintf(0, c, "recursive process active now!\n");

       return(0);

     }

-    return(mexec_scan(c, fn, h, mode));

+    return(mexec_scan(c, fn, t, mode));

   }

   /*----- help -----*/

   if(!fn){

@@ -326,9 +328,9 @@ int mexec_send(mcomm *c, int n, int sync)

 	}

   /*----- send to address set -----*/

-  if(h){

+  if(t){

     m->sendto = 1;

-    memcpy(&(m->addr.sin_addr), &(h->ad), sizeof(m->addr.sin_addr));

+    memcpy(&(m->addr.sin_addr), &(t->ad), sizeof(m->addr.sin_addr));

   }

 	strcpy(m->fn, fn);

@@ -336,7 +338,8 @@ int mexec_send(mcomm *c, int n, int sync)

 	m->mdata.head.opcode = MAKUO_OP_SEND;

   m->mdata.head.nstate = MAKUO_SENDSTATE_STAT;

 	m->comm      = c;

-  m->dryrun    = (mode == MAKUO_MEXEC_DRY);

+  m->dryrun    = dryrun;

+  m->recurs    = recurs;

   m->initstate = 1;

   if(m->dryrun){

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

@@ -396,8 +399,10 @@ int mexec_send(mcomm *c, int n, int sync)

 		  cprintf(0, c, "error: readlink error %s\n", fn);

 		  lprintf(0, "%s: readlink error %s\n", __func__, fn);

 		  mfdel(m);

+      return(0);

     }

-  }  

+  } 

+ 

   return(0);

 }

@@ -879,7 +884,8 @@ int mexec(mcomm *c, int n)

   if(n == 1){

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

       if(m->comm == c){

-        if(count++ == MAKUO_PARALLEL_MAX){

+        count++;

+        if(count == MAKUO_PARALLEL_MAX){

           return(-1);

         }

       }

@@ -339,6 +339,10 @@ static void minit_socket()

     lprintf(0, "%s: can't create multicast socket\n", __func__);

     exit(1);

   }

+  if(fcntl(s, F_SETFL , O_NONBLOCK)){

+    lprintf(0, "%s: fcntl error\n", __func__);

+    exit(1);

+  }

   if(bind(s, (struct sockaddr*)&addr, sizeof(addr)) == -1){

     lprintf(0, "%s: bind error\n", __func__);

     exit(1);

@@ -76,7 +76,10 @@ static int mrecv_packet(int s, mdata *data, struct sockaddr_in *addr)

     if(recvsize != -1){

       break;

     }else{

-      if(errno == EAGAIN || errno == EINTR){

+      if(errno == EAGAIN){

+        return(-1);

+      }

+      if(errno == EINTR){

         continue;

       }else{

         lprintf(0, "%s: recv error from %s\n", __func__, inet_ntoa(addr->sin_addr));

@@ -112,18 +115,19 @@ static int mrecv_packet(int s, mdata *data, struct sockaddr_in *addr)

 * Receive common functions (public)

 *

 *******************************************************************/

-void mrecv(int s)

+int mrecv(int s)

 {

   mdata  data;

   struct sockaddr_in addr;

   if(mrecv_packet(s, &data, &addr) == -1){

-    return;

+    return(0);

   }

   if(data.head.flags & MAKUO_FLAG_ACK){

     mrecv_ack(&data, &addr);

   }else{

     mrecv_req(&data, &addr);

   }

+  return(1);

 }

 void mrecv_gc()

@@ -1201,7 +1205,7 @@ static void mrecv_req_dsync_break(mfile *m, mdata *data, struct sockaddr_in *add

  */

 static void mrecv_req_dsync(mdata *data, struct sockaddr_in *addr)

 {

-  lprintf(9, "%s:\n", __func__);

+  lprintf(9, "%s: rid=%06d %s %s\n", __func__, data->head.reqid, OPCODE(data->head.opcode), SSTATE(data->head.nstate));

   mfile *m = mrecv_req_search(data, addr);

   switch(data->head.nstate){

     case MAKUO_SENDSTATE_OPEN:

@@ -201,12 +201,13 @@ static void msend_ack_md5(int s, mfile *m)

 static void msend_ack_dsync(int s, mfile *m)

 {

+  mprintf(__func__, m);

   msend_shot(s, m);

 }

 static void msend_ack_del(int s, mfile *m)

 {

-  lprintf(9, "%s:\n", __func__);

+  mprintf(__func__, m);

   msend_shot(s, m);

 }

@@ -843,7 +844,7 @@ static void msend_req_dsync_break(int s, mfile *m)

 /*----- dsync -----*/

 static void msend_req_dsync(int s, mfile *m)

 {

-  lprintf(9, "%s:\n", __func__);

+  lprintf(9, "%s: rid=%d %s %s\n", __func__, m->mdata.head.reqid, OPCODE(m->mdata.head.opcode), SSTATE(m->mdata.head.nstate));

   if(m->mdata.head.nstate != MAKUO_SENDSTATE_LAST){

     if(!m->comm){

       if(m->mdata.head.nstate != MAKUO_SENDSTATE_BREAK){

@@ -872,10 +873,24 @@ static void msend_req_del_mark(int s, mfile *m)

 {

   lprintf(9, "%s:\n", __func__);

   mfile *d = m->link; /* dsync object */

-  if(member_get(&(d->addr.sin_addr))){

-    mkack(&(d->mdata), &(d->addr), MAKUO_RECVSTATE_CLOSE);

-  }else{

-    d->lastrecv.tv_sec = 1;

+  if(m->initstate){

+    m->initstate = 0;

+    m->sendwait  = 1;

+    ack_clear(m, -1);

+    if(member_get(&(d->addr.sin_addr))){

+      mkack(&(d->mdata), &(d->addr), MAKUO_RECVSTATE_CLOSE);

+    }else{

+      d->lastrecv.tv_sec = 1;

+    }

+    return;

+  }

+  if(m->sendwait){

+    if(member_get(&(d->addr.sin_addr))){

+      mkack(&(d->mdata), &(d->addr), MAKUO_RECVSTATE_CLOSE);

+    }else{

+      d->lastrecv.tv_sec = 1;

+    }

+    return;

   }

 }

@@ -942,10 +957,12 @@ static void msend_req_del_stat(int s, mfile *m)

 static void msend_req_del_last(int s, mfile *m)

 {

+  lprintf(9, "%s:\n", __func__);

   msend_mfdel(m);

 }

 static void msend_req_del_break(int s, mfile *m)

 {

+  lprintf(9, "%s:\n", __func__);

   if(m->link){

     m->link->lastrecv.tv_sec = 1;

   }

@@ -954,6 +971,7 @@ static void msend_req_del_break(int s, mfile *m)

 static void msend_req_del_open(int s, mfile *m)

 {

+  lprintf(9, "%s:\n", __func__);

   if(m->initstate){

     m->initstate = 0;

     m->sendwait  = 1;

@@ -970,6 +988,7 @@ static void msend_req_del_open(int s, mfile *m)

 static void msend_req_del_data(int s, mfile *m)

 {

+  lprintf(9, "%s:\n", __func__);

   if(m->initstate){

     m->initstate = 0;

     m->sendwait  = 1;

@@ -988,6 +1007,7 @@ static void msend_req_del_data(int s, mfile *m)

 static void msend_req_del_close(int s, mfile *m)

 {

+  lprintf(9, "%s:\n", __func__);

   if(m->initstate){

     m->initstate = 0;

     m->sendwait  = 1;