add atomic refcnt and locks.

It might be unnecessary, if open/close can be done without lock.

4 files changed, 131 insertions, 13 deletions

diff --git a/src/atomic.h b/src/atomic.h
new file mode 100644
index 0000000..a840653
--- /dev/null
+++ b/src/atomic.h
@@ -0,0 +1,58 @@
+#ifndef _ATOMIC_H_
+#define _ATOMIC_H_
+
+#include <stdlib.h>
+#include <pthread.h>
+#include <util.h>
+
+typedef int refcnt;
+
+static inline void refcnt_inc(refcnt *cnt)
+{
+        __sync_add_and_fetch(cnt, 1);
+}
+
+static inline void refcnt_dec(refcnt *cnt)
+{
+        __sync_sub_and_fetch(cnt, 1);
+}
+
+
+typedef pthread_mutex_t lock;
+
+static inline void lock_init(lock *l)
+{
+        pthread_mutex_init(l, NULL);
+}
+
+static inline void lock_acquire(lock *l)
+{
+        int ret = pthread_mutex_lock(l);
+        if (ret < 0) {
+                switch (ret) {
+                case EINVAL:
+                        pr_err("invalid mutex\n");
+                        exit(1);
+                case EDEADLK:
+                        pr_err("a deadlock would occur\n");
+                        exit(1);
+                }
+        }
+}
+
+static inline void lock_release(lock *l)
+{
+        int ret = pthread_mutex_unlock(l);
+        if (ret < 0) {
+                switch (ret) {
+                case EINVAL:
+                        pr_err("invalid mutex\n");
+                        exit(1);
+                case EPERM:
+                        pr_err("this thread does not hold this mutex\n");
+                        exit(1);
+                }
+        }
+}
+
+#endif /* _ATOMIC_H_ */
diff --git a/src/file.c b/src/file.c
index 7f1c1d0..efe654e 100644
--- a/src/file.c
+++ b/src/file.c
@@ -117,6 +117,7 @@ static struct file *file_alloc(char *path, size_t size, bool remote)
 
         f->size = size;
         f->remote = remote;
+        lock_init(&f->lock);
 
         return f;
 }
@@ -269,11 +270,11 @@ int file_fill(sftp_session sftp, struct list_head *head, char **src_array, int c
 
 
 #ifdef DEBUG
-void file_dump(struct list_head *file_head)
+void file_dump(struct list_head *file_list)
 {
         struct file *f;
 
-        list_for_each_entry(f, file_head, list) {
+        list_for_each_entry(f, file_list, list) {
                 pr_debug("%s %s %lu-byte\n", f->path,
                          f->remote ? "(remote)" : "(local)", f->size);
         }
@@ -295,6 +296,7 @@ static void *chunk_alloc(struct file *f)
         c->f = f;
         c->off = 0;
         c->len = 0;
+        refcnt_inc(&f->refcnt);
         return c;
 }
 
@@ -312,7 +314,7 @@ static int get_page_mask(void)
         return page_mask >> 1;
 }
 
-int chunk_fill(struct list_head *file_head, struct list_head *chunk_head,
+int chunk_fill(struct list_head *file_list, struct list_head *chunk_list,
                int nr_conn, int min_chunk_sz, int max_chunk_sz)
 {
         struct chunk *c;
@@ -323,7 +325,7 @@ int chunk_fill(struct list_head *file_head, struct list_head *chunk_head,
 
         page_mask = get_page_mask();
 
-        list_for_each_entry(f, file_head, list) {
+        list_for_each_entry(f, file_list, list) {
                 if (f->size <= min_chunk_sz)
                         chunk_sz = f->size;
                 else if (max_chunk_sz)
@@ -344,7 +346,7 @@ int chunk_fill(struct list_head *file_head, struct list_head *chunk_head,
                         c->off = f->size - size;
                         c->len = size < chunk_sz ? size : chunk_sz;
                         size -= c->len;
-                        list_add_tail(&c->list, chunk_head);
+                        list_add_tail(&c->list, chunk_list);
                 }
         }
 
@@ -352,14 +354,30 @@ int chunk_fill(struct list_head *file_head, struct list_head *chunk_head,
 }
 
 #ifdef DEBUG
-void chunk_dump(struct list_head *chunk_head)
+void chunk_dump(struct list_head *chunk_list)
 {
         struct chunk *c;
 
-        list_for_each_entry(c, chunk_head, list) {
+        list_for_each_entry(c, chunk_list, list) {
                 pr_debug("%s %s 0x%010lx-0x%010lx %lu-byte\n",
                          c->f->path, c->f->remote ? "(remote)" : "(local)",
                          c->off, c->off + c->len, c->len);
         }
 }
 #endif
+
+
+struct chunk *chunk_acquire(struct list_head *chunk_list)
+{
+        /* under the lock for chunk_list */
+
+        struct list_head *first = chunk_list->next;
+        struct chunk *c = NULL;
+
+        if (list_empty(chunk_list))
+                return NULL; /* empty list */
+
+        c = list_entry(first, struct chunk, list);
+        list_del(first);
+        return c;
+}
diff --git a/src/file.h b/src/file.h
index 0c22944..9d79727 100644
--- a/src/file.h
+++ b/src/file.h
@@ -1,23 +1,59 @@
 #ifndef _FILE_H_
 #define _FILE_H_
 
+#include <pthread.h>
 #include <libssh/libssh.h>
 #include <libssh/sftp.h>
 #include <list.h>
+#include <atomic.h>
 
 struct file {
         struct list_head        list;   /* sscp->file_list */
 
-        char            *path;
-        bool            remote;
-        size_t          size;   /* size of this file */
+        char    *path;          /* copy source path */
+        char    *dst_path;      /* copy destination path */
+        bool    remote;
+        size_t  size;   /* size of this file */
+
+        int     state;  /* destination file state */
+        lock    lock;   /* mutex to protect state */
+        refcnt  refcnt; /* chunks referencing this file */
+
+        size_t  done;   /* copied bytes. a control thread totaling up done of chunks */
 };
+#define FILE_STATE_INIT         0
+#define FILE_STATE_OPENED       1
+#define FILE_STATE_DONE         2
+
+/* Allocating chunk increments refcnt of the associating file.
+ * Multiple threads copying files follows:
+ *
+ * acquire a chunk (inside a global lock)
+ *
+ * if the file state of the chunk is INIT:
+ *     acquire the file lock
+ * *         if file state is INIT:
+ *             create destination file
+ *             set file state OPENED.
+ *             // only the first thread in the lock open the destination file
+ *     release the file lock
+ * endif
+ *
+ * copy the chunk to the destination.
+ * decrement the refcnt of the file.
+ *
+ * if refcnt == 0:
+ *     all chunks are copied.
+ *     set the file state DONE, print something useful output.
+ * endif
+ */
 
 struct chunk {
         struct list_head        list;   /* sscp->chunk_list */
         struct file *f;
         size_t  off;    /* offset of this chunk on the file f */
         size_t  len;    /* length of this chunk */
+        size_t  done;   /* copied bytes for this chunk by a thread */
 };
 
 char *file_find_hostname(char *path);
@@ -26,12 +62,14 @@ int file_is_directory(char *path, sftp_session sftp);
 
 int file_fill(sftp_session sftp, struct list_head *head, char **src_array, int count);
 
-int chunk_fill(struct list_head *file_head, struct list_head *chunk_head,
+int chunk_fill(struct list_head *file_list, struct list_head *chunk_list,
                int nr_conn, int min_chunk_sz, int max_chunk_sz);
 
+struct chunk *chunk_acquire(struct list_head *chunk_list);
+
 #ifdef DEBUG
-void file_dump(struct list_head *file_head);
-void chunk_dump(struct list_head *chunk_head);
+void file_dump(struct list_head *file_list);
+void chunk_dump(struct list_head *chunk_list);
 #endif
 
 
diff --git a/src/main.c b/src/main.c
index 8f7f0fb..3f2211a 100644
--- a/src/main.c
+++ b/src/main.c
@@ -7,6 +7,7 @@
 #include <util.h>
 #include <ssh.h>
 #include <file.h>
+#include <atomic.h>
 #include <platform.h>
 
 int verbose = 0; /* util.h */
@@ -19,6 +20,8 @@ struct sscp {
 
         struct list_head        file_list;
         struct list_head        chunk_list;
+        lock                    chunk_lock;  /* lock for chunk list */
+
         char *target;
         bool target_is_remote;
 };
@@ -111,6 +114,7 @@ int main(int argc, char **argv)
         memset(&sscp, 0, sizeof(sscp));
         INIT_LIST_HEAD(&sscp.file_list);
         INIT_LIST_HEAD(&sscp.chunk_list);
+        lock_init(&sscp.chunk_lock);
 
 	while ((ch = getopt(argc, argv, "n:s:S:l:p:i:c:Cvh")) != -1) {
 		switch (ch) {