added Cache_background_download() to enable the download of the next block

Author: fangfufu

Makefile

@@ -1,7 +1,9 @@
 VERSION=1.1.0
 
-CFLAGS+= -g -O2 -Wall -Wextra -D_FILE_OFFSET_BITS=64 -DVERSION=\"$(VERSION)\" `pkg-config --cflags-only-I gumbo libcurl fuse`
-LDFLAGS+= -lgumbo -lcurl -lfuse -lcrypto `pkg-config --libs-only-L gumbo libcurl fuse`
+CFLAGS+= -g -O2 -Wall -Wextra -D_FILE_OFFSET_BITS=64 -DVERSION=\"$(VERSION)\" \
+`pkg-config --cflags-only-I gumbo libcurl fuse`
+LDFLAGS+= -pthread -lgumbo -lcurl -lfuse -lcrypto \
+`pkg-config --libs-only-L gumbo libcurl fuse`
 COBJS = main.o network.o fuse_local.o link.o cache.o util.o
 
 prefix ?= /usr/local

src/cache.c

@@ -45,11 +45,6 @@ typedef enum {
 
 int CACHE_SYSTEM_INIT = 0;
 
-/**
- * \brief the receive buffer
- */
-static uint8_t RECV_BUF[DATA_BLK_SZ];
-
 /**
  * \brief The metadata directory
  */
@@ -280,19 +275,13 @@ static long Data_size(const char *fn)
  *  - negative values on error,
  *  - otherwise, the number of bytes read.
  */
-static long Data_read(const Cache *cf, uint8_t *buf, off_t len, off_t offset)
+static long Data_read(Cache *cf, uint8_t *buf, off_t len, off_t offset)
 {
     if (len == 0) {
         fprintf(stderr, "Data_read(): requested to read 0 byte!\n");
         return -EINVAL;
     }
 
-    size_t start = offset;
-    size_t end = start + len;
-    char range_str[64];
-    snprintf(range_str, sizeof(range_str), "%lu-%lu", start, end);
-    fprintf(stderr, "Data_read(%s, %s);\n", cf->path, range_str);
-
     long byte_read = -EIO;
 
     if (fseeko(cf->dfp, offset, SEEK_SET)) {
@@ -332,20 +321,14 @@ static long Data_read(const Cache *cf, uint8_t *buf, off_t len, off_t offset)
  *  - otherwise, the number of bytes written.
  */
 
-static long Data_write(const Cache *cf, const uint8_t *buf, off_t len,
+static long Data_write(Cache *cf, const uint8_t *buf, off_t len,
                        off_t offset)
 {
     if (len == 0) {
         fprintf(stderr, "Data_write(): requested to write 0 byte!\n");
         return -EINVAL;
     }
 
-    size_t start = offset;
-    size_t end = start + len;
-    char range_str[64];
-    snprintf(range_str, sizeof(range_str), "%lu-%lu", start, end);
-    fprintf(stderr, "Data_write(%s, %s);\n", cf->path, range_str);
-
     long byte_written = -EIO;
 
     if (fseeko(cf->dfp, offset, SEEK_SET)) {
@@ -404,8 +387,26 @@ static Cache *Cache_alloc()
     }
 
     if (pthread_mutex_init(&cf->rw_lock, NULL)) {
-        printf(
-            "Cache_alloc(): rw_lock initialisation failed!\n");
+        fprintf(stderr, "Cache_alloc(): rw_lock initialisation failed!\n");
+    }
+
+    if (pthread_mutex_init(&cf->bgt_lock, NULL)) {
+        fprintf(stderr, "Cache_alloc(): seg_lock initialisation failed!\n");
+    }
+
+
+    if (pthread_mutexattr_init(&cf->bgt_lock_attr)) {
+        fprintf(stderr,
+                "Cache_alloc(): bgt_lock_attr initialisation failed!\n");
+    }
+
+    if (pthread_mutexattr_setpshared(&cf->bgt_lock_attr,
+        PTHREAD_PROCESS_SHARED)) {
+        fprintf(stderr, "Cache_alloc(): could not set bgt_lock_attr!\n");
+    }
+
+    if (pthread_mutex_init(&cf->bgt_lock, NULL)) {
+        fprintf(stderr, "Cache_alloc(): bgt_lock initialisation failed!\n");
     }
 
     return cf;
@@ -420,12 +421,21 @@ static void Cache_free(Cache *cf)
         fprintf(stderr, "Cache_free(): could not destroy rw_lock!\n");
     }
 
+    if (pthread_mutex_destroy(&cf->bgt_lock)) {
+        fprintf(stderr, "Cache_free(): could not destroy bgt_lock!\n");
+    }
+
+    if (pthread_mutexattr_destroy(&cf->bgt_lock_attr)) {
+        fprintf(stderr, "Cache_alloc(): could not destroy bgt_lock_attr!\n");
+    }
+
     if (cf->path) {
         free(cf->path);
     }
     if (cf->seg) {
         free(cf->seg);
     }
+
     free(cf);
 }
 
@@ -626,7 +636,6 @@ Cache *Cache_open(const char *fn)
 {
     /* Check if both metadata and data file exist */
     if (!Cache_exist(fn)) {
-//         fprintf(stderr, "dataset does not exist!\n");
         return NULL;
     }
 
@@ -689,6 +698,10 @@ cf->content_length: %ld, Data_size(fn): %ld.\n", fn, cf->content_length,
 
 void Cache_close(Cache *cf)
 {
+    /* Must wait for the background download thread to stop */
+    pthread_mutex_lock(&cf->bgt_lock);
+    pthread_mutex_unlock(&cf->bgt_lock);
+
     if (Meta_write(cf)) {
         fprintf(stderr, "Cache_close(): Meta_write() error.");
     }
@@ -728,9 +741,41 @@ static void Seg_set(Cache *cf, off_t offset, int i)
     cf->seg[byte] = i;
 }
 
+/**
+ * \brief Background download function
+ * \details If we are requesting the data from the second half of the current
+ * segment, we can spawn a pthread using this function to download the next
+ * segment.
+ */
+static void *Cache_background_download(void *arg)
+{
+    fprintf(stderr, "Starting Cache_background_download in its own thread.\n");
+    Cache *cf = (Cache *) arg;
+    uint8_t recv_buf[DATA_BLK_SZ];
+
+    long recv = path_download(cf->path, (char *) recv_buf, cf->blksz,
+                              cf->next_offset);
+    if ( (recv == cf->blksz) ||
+        (cf->next_offset == (cf->content_length / cf->blksz * cf->blksz)) )
+    {
+        Data_write(cf, recv_buf, cf->blksz, cf->next_offset);
+        Seg_set(cf, cf->next_offset, 1);
+    }  else {
+            fprintf(stderr,
+                    "Cache_background_download(): recv (%ld) < cf->blksz! \
+Possible network error?\n",
+                    recv);
+    }
+
+    pthread_mutex_unlock(&cf->bgt_lock);
+    fprintf(stderr, "Exiting Cache_background_download thread.\n");
+    pthread_exit(NULL);
+}
+
 long Cache_read(Cache *cf, char *output_buf, off_t len, off_t offset)
 {
     long send;
+    uint8_t recv_buf[DATA_BLK_SZ];
     /*
      * Quick fix for SIGFPE,
      * this shouldn't happen in the first place!
@@ -741,18 +786,19 @@ long Cache_read(Cache *cf, char *output_buf, off_t len, off_t offset)
                 cf->blksz);
         return path_download(cf->path, output_buf, len, offset);
     }
+
     pthread_mutex_lock(&cf->rw_lock);
+    /* Calculate the aligned offset */
+    off_t dl_offset = offset / cf->blksz * cf->blksz;
     if (Seg_exist(cf, offset)) {
         /*
          * The metadata shows the segment already exists. This part is easy,
          * as you don't have to worry about alignment
          */
         send = Data_read(cf, (uint8_t *) output_buf, len, offset);
     } else {
-        /* Calculate the aligned offset */
-        off_t dl_offset = offset / cf->blksz * cf->blksz;
         /* Download the segment */
-        long recv = path_download(cf->path, (char *) RECV_BUF, cf->blksz,
+        long recv = path_download(cf->path, (char *) recv_buf, cf->blksz,
                                   dl_offset);
         /*
          * check if we have received enough data
@@ -762,19 +808,34 @@ long Cache_read(Cache *cf, char *output_buf, off_t len, off_t offset)
          * Condition 2: offset is the last segment
          */
         if ( (recv == cf->blksz) ||
-            (dl_offset == (cf->content_length / cf->blksz * cf->blksz)) ) {
-            memmove(output_buf, RECV_BUF + (offset - dl_offset), len);
+            (dl_offset == (cf->content_length / cf->blksz * cf->blksz)) )
+        {
+            memmove(output_buf, recv_buf + (offset - dl_offset), len);
             send = len;
-            Data_write(cf, RECV_BUF, cf->blksz, dl_offset);
+            Data_write(cf, recv_buf, cf->blksz, dl_offset);
             Seg_set(cf, dl_offset, 1);
         }  else {
-            memmove(output_buf, RECV_BUF + (offset - dl_offset), recv);
+            memmove(output_buf, recv_buf + (offset - dl_offset), recv);
             send = recv;
             fprintf(stderr,
             "Cache_read(): recv (%ld) < cf->blksz! Possible network error?\n",
                 recv);
         }
     }
     pthread_mutex_unlock(&cf->rw_lock);
+
+    /* Download the next segment in background */
+    cf->next_offset = round_div(offset, cf->blksz) * cf->blksz;
+    if ( (cf->next_offset > dl_offset) && !Seg_exist(cf, cf->next_offset) ) {
+        /* Stop the spawning of multiple background pthreads */
+        if(!pthread_mutex_trylock(&cf->bgt_lock)) {
+            if (pthread_create(&cf->bgt, NULL, Cache_background_download, cf)) {
+                fprintf(stderr,
+                    "Cache_read(): Error creating background download thread\n"
+                );
+            }
+        }
+    }
+
     return send;
 }

src/cache.h

@@ -24,12 +24,19 @@ typedef uint8_t Seg;
  */
 typedef struct {
     char *path; /**< the path to the file on the web server */
+    Link *link; /**< The Link associated with this cache data set */
     long time; /**<the modified time of the file */
     off_t content_length; /**<the size of the file */
-    pthread_mutex_t rw_lock; /**< mutex for disk operation */
+
+    pthread_t bgt; /**< background pthread */
+    pthread_mutex_t bgt_lock; /**< mutex for spawning a background thread */
+    pthread_mutexattr_t bgt_lock_attr;
+    off_t next_offset; /**<the offset of the next segment to be
+    downloaded in background*/
+
+    pthread_mutex_t rw_lock; /**< mutex for read/write operation */
     FILE *dfp; /**< The FILE pointer for the data file*/
     FILE *mfp; /**< The FILE pointer for the metadata */
-    Link *link; /**< The Link associated with this cache data set */
     int blksz; /**<the block size of the data file */
     long segbc; /**<segment array byte count */
     Seg *seg; /**< the detail of each segment */

src/util.c

@@ -28,3 +28,8 @@ char *strndupcat(const char *a, const char *b, int n)
     c[nc-1] = '\0';
     return c;
 }
+
+int64_t round_div(int64_t a, int64_t b)
+{
+    return (a + (b / 2)) / b;
+}

src/util.h

@@ -1,6 +1,8 @@
 #ifndef UTIL_H
 #define UTIL_H
 
+#include <stdint.h>
+
 /**
  * \file util.h
  * \brief utility functions
@@ -16,4 +18,9 @@
  */
 char *strndupcat(const char *a, const char *b, int n);
 
+/**
+ * \brief division, but rounded to the nearest integer rather than truncating
+ */
+int64_t round_div(int64_t a, int64_t b);
+
 #endif