stored: Implement object device chunked volumes.

As some of the REST protocols only allow you to write blobs in
total we now offer a chunked object store device type in which
the volumes are chunked into pieces and these pieces are flushed
to the backing store either on close or when we reach the maximum
size of the chunk and need a new one. We also implemented the
reading of such chunked devices.

This runs on top of Scality's libdroplet with some local fixes and
changes.

autoconf/configure.in

@@ -888,9 +888,10 @@ AC_HEADER_DIRENT
 AC_CHECK_HEADER(glob.h, [AC_DEFINE(HAVE_GLOB_H, 1, [Define to 1 if you have the <glob.h> header file.])] , )
 AC_CHECK_HEADER(poll.h, [AC_DEFINE(HAVE_POLL_H, 1, [Define to 1 if you have the <poll.h> header file.])] , )
 AC_CHECK_HEADER(sys/poll.h, [AC_DEFINE(HAVE_SYS_POLL_H, 1, [Define to 1 if you have the <sys/poll.h> header file.])] , )
+AC_CHECK_HEADER(sys/mman.h, [AC_DEFINE(HAVE_SYS_MMAN_H, 1, [Define to 1 if you have the <sys/mman.h> header file.])] , )
 AC_CHECK_FUNCS(glob strcasecmp select poll setenv putenv tcgetattr)
 AC_CHECK_FUNCS(lstat lchown lchmod utimes lutimes futimes futimens fchmod fchown)
-AC_CHECK_FUNCS(nanosleep nl_langinfo)
+AC_CHECK_FUNCS(mmap nanosleep nl_langinfo)
 AC_CHECK_HEADERS(varargs.h)
 
 
@@ -4542,14 +4543,18 @@ AC_SUBST_FILE(DEBIAN_CONTROL_STORAGE_PYTHON_PLUGIN)
 AC_SUBST_FILE(DEBIAN_CONTROL_DIRECTOR_PYTHON_PLUGIN)
 
 dnl build a list of storage backends we need to build.
+BUILD_SD_BACKENDS=""
 if test x$use_libtool != xno; then
-   BUILD_SD_BACKENDS="libbareossd-fifo.la libbareossd-gentape.la libbareossd-tape.la"
+   BUILD_SD_BACKENDS="${BUILD_SD_BACKENDS} libbareossd-fifo.la"
+   BUILD_SD_BACKENDS="${BUILD_SD_BACKENDS} libbareossd-gentape.la"
+   BUILD_SD_BACKENDS="${BUILD_SD_BACKENDS} libbareossd-tape.la"
 
    if test X"$have_glusterfs" = "Xyes" ; then
       BUILD_SD_BACKENDS="${BUILD_SD_BACKENDS} libbareossd-gfapi.la"
    fi
 
    if test X"$have_droplet" = "Xyes" ; then
+      BUILD_SD_BACKENDS="${BUILD_SD_BACKENDS} libbareossd-chunked.la"
       BUILD_SD_BACKENDS="${BUILD_SD_BACKENDS} libbareossd-object.la"
    fi
 
@@ -4564,13 +4569,10 @@ if test x$use_libtool != xno; then
    if test X"$have_elasto" = "Xyes" ; then
       BUILD_SD_BACKENDS="${BUILD_SD_BACKENDS} libbareossd-elasto.la"
    fi
-else
-   BUILD_SD_BACKENDS=""
 fi
 
 AC_SUBST(BUILD_SD_BACKENDS)
 
-
 dnl Insanity check
 if test "x${subsysdir}" = "x${sbindir}" ; then
    echo " "

src/lib/Makefile.in

@@ -44,11 +44,11 @@ INCLUDE_FILES = ../include/baconfig.h ../include/bareos.h \
 		bsock_tcp.h bsock_udt.h bsr.h btime.h btimers.h cbuf.h \
 		crypto.h crypto_cache.h devlock.h dlist.h fnmatch.h \
 		guid_to_name.h htable.h ini.h lex.h lib.h lockmgr.h \
-		md5.h mem_pool.h message.h mntent_cache.h parse_conf.h \
-		plugins.h protos.h queue.h rblist.h runscript.h rwlock.h \
-		scsi_crypto.h scsi_lli.h scsi_tapealert.h sellist.h \
-		serial.h sha1.h smartall.h status.h tls.h tree.h var.h \
-		waitq.h watchdog.h workq.h
+		md5.h mem_pool.h message.h mntent_cache.h ordered_cbuf.h \
+		parse_conf.h plugins.h protos.h queue.h rblist.h \
+		runscript.h rwlock.h scsi_crypto.h scsi_lli.h \
+		scsi_tapealert.h sellist.h serial.h sha1.h smartall.h \
+		status.h tls.h tree.h var.h waitq.h watchdog.h workq.h
 
 #
 # libbareos
@@ -61,7 +61,7 @@ LIBBAREOS_SRCS = address_conf.c alist.c attr.c attribs.c base64.c \
 		 crypto_cache.c crypto_gnutls.c crypto_none.c crypto_nss.c \
 		 crypto_openssl.c crypto_wrap.c daemon.c devlock.c dlist.c \
 		 edit.c fnmatch.c guid_to_name.c hmac.c htable.c jcr.c json.c \
-		 lockmgr.c md5.c mem_pool.c message.c mntent_cache.c \
+		 lockmgr.c md5.c mem_pool.c message.c mntent_cache.c ordered_cbuf.c \
 		 output_formatter.c passphrase.c path_list.c plugins.c poll.c \
 		 priv.c queue.c rblist.c runscript.c rwlock.c scan.c scsi_crypto.c \
 		 scsi_lli.c scsi_tapealert.c sellist.c serial.c sha1.c signal.c \

src/lib/cbuf.c

@@ -31,7 +31,7 @@
 /*
  * Initialize a new circular buffer.
  */
-int circbuf::init()
+int circbuf::init(int capacity)
 {
    if (pthread_mutex_init(&m_lock, NULL) != 0) {
       return -1;
@@ -51,7 +51,11 @@ int circbuf::init()
    m_next_in = 0;
    m_next_out = 0;
    m_size = 0;
-   m_capacity = QSIZE;
+   m_capacity = capacity;
+   if (m_data) {
+      free(m_data);
+   }
+   m_data = (void **)malloc(m_capacity * sizeof(void *));
 
    return 0;
 }
@@ -64,6 +68,10 @@ void circbuf::destroy()
    pthread_cond_destroy(&m_notempty);
    pthread_cond_destroy(&m_notfull);
    pthread_mutex_destroy(&m_lock);
+   if (m_data) {
+      free(m_data);
+      m_data = NULL;
+   }
 }
 
 /*
@@ -86,9 +94,9 @@ int circbuf::enqueue(void *data)
    m_next_in %= m_capacity;
 
    /*
-    * Let a waiting consumer know there is data.
+    * Let any waiting consumer know there is data.
     */
-   pthread_cond_signal(&m_notempty);
+   pthread_cond_broadcast(&m_notempty);
 
    pthread_mutex_unlock(&m_lock);
 
@@ -100,7 +108,7 @@ int circbuf::enqueue(void *data)
  */
 void *circbuf::dequeue()
 {
-   void *data;
+   void *data = NULL;
 
    if (pthread_mutex_lock(&m_lock) != 0) {
       return NULL;
@@ -117,21 +125,19 @@ void *circbuf::dequeue()
     * When we are requested to flush and there is no data left return NULL.
     */
    if (empty() && m_flush) {
-      m_flush = false;
-      pthread_mutex_unlock(&m_lock);
-
-      return NULL;
+      goto bail_out;
    }
 
    data = m_data[m_next_out++];
    m_size--;
    m_next_out %= m_capacity;
 
    /*
-    * Let a waiting producer know there is room.
+    * Let all waiting producers know there is room.
     */
-   pthread_cond_signal(&m_notfull);
+   pthread_cond_broadcast(&m_notfull);
 
+bail_out:
    pthread_mutex_unlock(&m_lock);
 
    return data;
@@ -169,12 +175,15 @@ int circbuf::flush()
       return -1;
    }
 
+   /*
+    * Set the flush flag.
+    */
    m_flush = true;
 
    /*
-    * Let a waiting consumer know there will be no more data.
+    * Let all waiting consumers know there will be no more data.
     */
-   pthread_cond_signal(&m_notempty);
+   pthread_cond_broadcast(&m_notempty);
 
    pthread_mutex_unlock(&m_lock);
 

src/lib/cbuf.h

@@ -29,6 +29,7 @@
 #define QSIZE 10              /* # of pointers in the queue */
 
 class circbuf : public SMARTALLOC {
+private:
    int m_size;
    int m_next_in;
    int m_next_out;
@@ -37,28 +38,30 @@ class circbuf : public SMARTALLOC {
    pthread_mutex_t m_lock;    /* Lock the structure */
    pthread_cond_t m_notfull;  /* Full -> not full condition */
    pthread_cond_t m_notempty; /* Empty -> not empty condition */
-   void *m_data[QSIZE];       /* Circular buffer of pointers */
+   void **m_data;             /* Circular buffer of pointers */
 
 public:
-   circbuf();
+   circbuf(int capacity = QSIZE);
    ~circbuf();
-   int init();
+   int init(int capacity);
    void destroy();
    int enqueue(void *data);
    void *dequeue();
    int next_slot();
    int flush();
    bool full() { return m_size == m_capacity; };
    bool empty() { return m_size == 0; };
+   bool is_flushing() { return m_flush; };
    int capacity() const { return m_capacity; };
 };
 
 /*
  * Constructor
  */
-inline circbuf::circbuf()
+inline circbuf::circbuf(int capacity)
 {
-   init();
+   m_data = NULL;
+   init(capacity);
 }
 
 /*