@@ -271,7 +271,7 @@ int seq_addmark(mfile *m, uint32_t lseq, uint32_t useq)

   if(!m->mark){

     m->markcount = 0;

-    m->marksize  = 512;

+    m->marksize  = 1024;

     m->mark = malloc(sizeof(uint32_t) * m->marksize);

     if(!m->mark){

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

@@ -280,7 +280,7 @@ int seq_addmark(mfile *m, uint32_t lseq, uint32_t useq)

   }

   size = m->marksize;

   while(size < m->markcount + useq - lseq)

-    size += 512;

+    size += 1024;

   if(size != m->marksize){

     n = realloc(m->mark, sizeof(uint32_t) * size);

     if(!n){

@@ -296,6 +296,7 @@ int seq_addmark(mfile *m, uint32_t lseq, uint32_t useq)

       if(!a){

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

       }else{

+        lprintf(0,"%s: mark over(%d/%d) retry %s\n", __func__, m->markcount, m->marksize, m->fn);

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

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

         a->mdata.head.reqid  = m->mdata.head.reqid;

@@ -350,7 +351,6 @@ void clr_hoststate(mfile *m)

 {

   int    i;

   mhost *t;

-

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

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

       if(t->mflist[i] == m){

@@ -361,6 +361,17 @@ void clr_hoststate(mfile *m)

   }

 }

+void dump_hoststate(mfile *m, char *func)

+{

+  mhost   *t;

+  uint8_t *r;

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

+    if(r=get_hoststate(t,m)){

+      lprintf(9,"%s: state=%d from %s %s\n", func, (int)(*r), t->hostname, m->fn);

+    }

+  }

+}

+

 uint8_t *get_hoststate(mhost *t, mfile *m)

 {

   int i;

@@ -81,18 +81,20 @@ int restoreguid()

 void recv_timeout(mfile *m)

 {

-  mhost *h;

+  mhost   *t;

+  uint8_t *r;

   if(m){

     m->retrycnt = MAKUO_SEND_RETRYCNT;

     do{

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

-        if(h->state == MAKUO_RECVSTATE_NONE){

-          lprintf(0,"%s: %s(%s) timeout\n", __func__, inet_ntoa(h->ad), h->hostname);

-          member_del(h);

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

+        r = get_hoststate(t, m);

+        if(*r == MAKUO_RECVSTATE_NONE){

+          lprintf(0,"%s: %s(%s) timeout\n", __func__, inet_ntoa(t->ad), t->hostname);

+          member_del(t);

           break;

         }

       }

-    }while(h); 

+    }while(t); 

   }

 }

@@ -254,44 +256,45 @@ int mcomm_fdset(mcomm *c, fd_set *fds)

   return(0);

 }

-int ismsend(mfile *m)

+int ismsend(int s, mfile *m)

 {

   int r;

-

-  if(!m)

+  if(!m){

     return(0);

-  if(!m->sendwait){

-    return(1);

   }

-  r = ack_check(m,MAKUO_RECVSTATE_NONE);

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

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

+      return(0);

+    }

+  }

+  if(m->senddelay){

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

+      return(1);

+    }

+  }

+  r = ack_check(m, MAKUO_RECVSTATE_NONE);

   if(r == -1){

     m->mdata.head.seqno  = 0;

     m->mdata.head.nstate = MAKUO_SENDSTATE_BREAK;

-    return(1);

   }

   if(!r){

     m->sendwait = 0;

-    if(!(m->senddelay)){

-      return(1);

-    }

-    if(mtimeout(&(m->lastsend), m->senddelay)){

-      return(1);

-    }

   }

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

-    if(m->retrycnt){

+  if(m->sendwait){

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

       return(1);

-    }else{

+    }

+    if(!(m->retrycnt)){

       recv_timeout(m);

     }

   }

-  return(0);

+  msend(s,m);

+  return(1);

 }

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

 int mloop()

 {

-  int i;

   fd_set rfds;

   fd_set wfds;

   struct timeval *lastpong;

@@ -321,23 +324,14 @@ int mloop()

     gettimeofday(&curtime,NULL);

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

-      mfile *m = mftop[0];

+      int para = 0;

       mfile *n = NULL;

-      for(i=0;i<moption.parallel;i++){

+      mfile *m = mftop[0];

+      while(m){

         n = m->next;

-        if(ismsend(m)){

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

-            msend(moption.mcsocket, m);

-          }else{

-            if(m == mftop[0]){

-              msend(moption.mcsocket, m);

-            }else{

-              i--;

-            }

-          }

-        }

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

         m = n;

-        if(!m){

+        if(para == moption.parallel){

           break;

         }

       }

@@ -202,10 +202,9 @@ static void mrecv_ack_ping(mdata *data, struct sockaddr_in *addr)

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

 {

-  uint8_t *s;

+  uint8_t *r;

   mhost   *t;

   mfile   *m;

-

   if(mrecv_ack_search(&t, &m, data, addr)){

     return;

   }

@@ -215,7 +214,7 @@ static void mrecv_ack_send(mdata *data, struct sockaddr_in *addr)

     lprintf(0,          "%s: file update ignore rid=%06d state=%02d %s(%s) %s\n", __func__, 

       data->head.reqid, data->head.nstate, inet_ntoa(t->ad), t->hostname, m->fn);

   }

-  if(data->head.nstate == MAKUO_RECVSTATE_OPEN){

+  if(data->head.nstate == MAKUO_RECVSTATE_MARK){

     uint32_t *d = (uint32_t *)(data->data);

     while(d < (uint32_t *)&data->data[data->head.szdata]){

       if(*d >= m->seqnomax){

@@ -235,8 +234,8 @@ static void mrecv_ack_send(mdata *data, struct sockaddr_in *addr)

     m->mdata.head.seqno  = 0;

     m->mdata.head.nstate = MAKUO_SENDSTATE_DATA;

   }else{

-    if(s = get_hoststate(t, m)){

-      *s = data->head.nstate;

+    if(r = get_hoststate(t, m)){

+      *r = data->head.nstate;

     }else{

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

     }

@@ -249,7 +248,6 @@ static void mrecv_ack_md5(mdata *data, struct sockaddr_in *addr)

   uint8_t *s;

   mhost   *t;

   mfile   *m;

-

   mrecv_ack_search(&t, &m, data, addr);

   if(!t || !m){

     return;

@@ -286,7 +284,6 @@ static void mrecv_ack(mdata *data, struct sockaddr_in *addr)

     case MAKUO_OP_MD5:

       mrecv_ack_md5(data, addr);

       break;

-

     /* 機能追加はここへ */

   }

 }

@@ -345,7 +342,7 @@ static void mrecv_req_send_break(mfile *m, mdata *r)

 static void mrecv_req_send_stat(mfile *m, mdata *r)

 {

-  mfile *a;

+  mfile  *a;

   struct stat fs;

   struct utimbuf mftime;

@@ -500,8 +497,10 @@ static void mrecv_req_send_mark(mfile *m, mdata *r)

   a->mdata.head.opcode = r->head.opcode;

   a->mdata.head.reqid  = r->head.reqid;

   a->mdata.head.seqno  = r->head.seqno;

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

+  a->mdata.head.ostate = m->mdata.head.nstate;

+  a->mdata.head.nstate = MAKUO_RECVSTATE_MARK;

   a->mdata.head.szdata = 0;

+

   memcpy(&(a->addr), &(m->addr), sizeof(a->addr));

   m->lickflag = 1;

   if(m->mdata.head.seqno < m->seqnomax){

@@ -292,6 +292,7 @@ static void msend_req_send_open_init(int s, mfile *m)

   m->sendwait  = 1;

   m->initstate = 0;

   ack_clear(m, MAKUO_RECVSTATE_UPDATE);

+

   /*----- symlink -----*/

   if(S_ISLNK(m->fs.st_mode)){

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

@@ -326,23 +327,23 @@ static void msend_req_send_open(int s, mfile *m)

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

     return;

   }

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

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

-    m->sendwait  = 1;

+  lprintf(9, "%s: %s\n", __func__, m->fn);

+  if(ack_check(m, MAKUO_RECVSTATE_UPDATE) == 1){

+    m->sendwait = 1;

     ack_clear(m, MAKUO_RECVSTATE_UPDATE);

+    return;

+  }

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

+    m->initstate = 1;

+    m->mdata.head.nstate = MAKUO_SENDSTATE_CLOSE;

   }else{

-    if(S_ISLNK(m->fs.st_mode)){

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

       m->initstate = 1;

       m->mdata.head.nstate = MAKUO_SENDSTATE_CLOSE;

-    }else{

-      if(S_ISDIR(m->fs.st_mode)){

-        m->initstate = 1;

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

-      }

-      if(S_ISREG(m->fs.st_mode)){

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

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

-      }

+    }

+    if(S_ISREG(m->fs.st_mode)){

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

+      m->mdata.head.nstate = MAKUO_SENDSTATE_DATA;

     }

   }

 }

@@ -351,6 +352,7 @@ static void msend_req_send_markdata(int s, mfile *m)

 {

   int i;

   int r;

+  dump_hoststate(m, __func__);

   if(!m->markcount){

     /* close */

     m->initstate = 1;

@@ -375,11 +377,11 @@ static void msend_req_send_markdata(int s, mfile *m)

       }

     }

   }

-  /* eof */

   m->markcount = 0;

   m->initstate = 1;

   m->mdata.head.seqno  = 0;

   m->mdata.head.nstate = MAKUO_SENDSTATE_MARK;

+  ack_clear(m, MAKUO_RECVSTATE_MARK);

 }

 static void msend_req_send_filedata(int s, mfile *m)

@@ -395,10 +397,10 @@ static void msend_req_send_filedata(int s, mfile *m)

   }else{

     if(readsize == -1){

       /* err */

-      lprintf(0, "%s: read error! seqno=%d errno=%d\n", __func__, m->mdata.head.seqno, errno);

+      lprintf(0, "%s: read error! seqno=%d errno=%d %s\n", __func__, m->mdata.head.seqno, errno, m->fn);

     }else{

       /* eof */

-      lprintf(4, "%s: block send count=%d\n", __func__, m->mdata.head.seqno);

+      lprintf(4, "%s: block send count=%d %s\n", __func__, m->mdata.head.seqno, m->fn);

       m->mdata.head.seqno  = 0;

       m->mdata.head.nstate = MAKUO_SENDSTATE_MARK;

       m->initstate = 1;

@@ -425,6 +427,7 @@ static void msend_req_send_mark_init(int s, mfile *m)

 {

   m->sendwait  = 1;

   m->initstate = 0;

+  ack_clear(m, MAKUO_RECVSTATE_UPDATE);

   ack_clear(m, MAKUO_RECVSTATE_OPEN);

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

 }

@@ -439,6 +442,15 @@ static void msend_req_send_mark(int s, mfile *m)

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

     return;

   }

+  if(ack_check(m, MAKUO_RECVSTATE_UPDATE) == 1){

+    msend_req_send_mark_init(s, m);

+    return;

+  }

+  if(ack_check(m, MAKUO_RECVSTATE_OPEN) == 1){

+    msend_req_send_mark_init(s, m);

+    return;

+  }

+  lprintf(9, "%s: mark=%d %s\n", __func__, m->markcount, m->fn);

   m->mdata.head.nstate = MAKUO_SENDSTATE_DATA;

 }

@@ -446,8 +458,9 @@ static void msend_req_send_close_init(int s, mfile *m)

 {

   m->sendwait  = 1;

   m->initstate = 0;

-  ack_clear(m, MAKUO_RECVSTATE_OPEN);

   ack_clear(m, MAKUO_RECVSTATE_UPDATE);

+  ack_clear(m, MAKUO_RECVSTATE_OPEN);

+  ack_clear(m, MAKUO_RECVSTATE_MARK);

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

 }

@@ -461,14 +474,16 @@ static void msend_req_send_close(int s, mfile *m)

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

     return;

   }

+  if(ack_check(m, MAKUO_RECVSTATE_UPDATE) == 1){

+    msend_req_send_close_init(s, m);

+    return;

+  }

   if(ack_check(m, MAKUO_RECVSTATE_OPEN) == 1){

-    m->sendwait  = 1;

-    ack_clear(m, MAKUO_RECVSTATE_OPEN);

+    msend_req_send_close_init(s, m);

     return;

   }

-  if(ack_check(m, MAKUO_RECVSTATE_UPDATE) == 1){

-    m->sendwait  = 1;

-    ack_clear(m, MAKUO_RECVSTATE_UPDATE);

+  if(ack_check(m, MAKUO_RECVSTATE_MARK) == 1){

+    msend_req_send_close_init(s, m);

     return;

   }

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