text: /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
  
  /*  Chunk I/O
   *  =========
   *  Copyright 2018 Eduardo Silva <eduardo@monkey.io>
   *
   *  Licensed under the Apache License, Version 2.0 (the "License");
   *  you may not use this file except in compliance with the License.
   *  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   *  Unless required by applicable law or agreed to in writing, software
   *  distributed under the License is distributed on an "AS IS" BASIS,
   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   *  See the License for the specific language governing permissions and
   *  limitations under the License.
   */
  
  #include <chunkio/chunkio_compat.h>
  #include <chunkio/chunkio.h>
  #include <chunkio/cio_version.h>
  #include <chunkio/cio_file.h>
  #include <chunkio/cio_memfs.h>
  #include <chunkio/cio_log.h>
  
  #include <string.h>
  
  struct cio_chunk *cio_chunk_open(struct cio_ctx *ctx, struct cio_stream *st,
                                   const char *name, int flags, size_t size,
                                   int *err)
  {
      int len;
      void *backend = NULL;
      struct cio_chunk *ch;
  
      if (!st) {
          cio_log_error(ctx, "[cio chunk] invalid stream");
          return NULL;
      }
  
      if (!name) {
          cio_log_error(ctx, "[cio chunk] invalid file name");
          return NULL;
      }
  
      len = strlen(name);
      if (len == 0) {
          cio_log_error(ctx, "[cio chunk] invalid file name");
          return NULL;
      }
  #ifndef CIO_HAVE_BACKEND_FILESYSTEM
      if (st->type == CIO_STORE_FS) {
          cio_log_error(ctx, "[cio chunk] file system backend not supported");
          return NULL;
      }
  #endif
  
      /* allocate chunk context */
      ch = malloc(sizeof(struct cio_chunk));
      if (!ch) {
          cio_errno();
          return NULL;
      }
      ch->name = strdup(name);
      ch->ctx = ctx;
      ch->st = st;
      ch->lock = CIO_FALSE;
      ch->tx_active = CIO_FALSE;
      ch->tx_crc = 0;
      ch->tx_content_length = 0;
      ch->backend = NULL;
  
      mk_list_add(&ch->_head, &st->chunks);
  
      /* create backend context */
      if (st->type == CIO_STORE_FS) {
          backend = cio_file_open(ctx, st, ch, flags, size, err);
      }
      else if (st->type == CIO_STORE_MEM) {
          *err = CIO_OK;
          backend = cio_memfs_open(ctx, st, ch, flags, size);
      }
  
      if (!backend) {
          mk_list_del(&ch->_head);
          free(ch->name);
          free(ch);
          return NULL;
      }
  
      ch->backend = backend;
  
      /* Adjust counter */
      cio_chunk_counter_total_add(ctx);
  
      /* Link the chunk state to the proper stream list */
      if (cio_chunk_is_up(ch) == CIO_TRUE) {
          mk_list_add(&ch->_state_head, &st->chunks_up);
      }
      else {
          mk_list_add(&ch->_state_head, &st->chunks_down);
      }
  
      return ch;
  }
  
  void cio_chunk_close(struct cio_chunk *ch, int delete)
  {
      int type;
      struct cio_ctx *ctx;
  
      if (!ch) {
          return;
      }
  
      ctx = ch->ctx;
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          cio_memfs_close(ch);
      }
      else if (type == CIO_STORE_FS) {
          cio_file_close(ch, delete);
      }
  
      mk_list_del(&ch->_head);
      mk_list_del(&ch->_state_head);
      free(ch->name);
      free(ch);
  
      /* Adjust counter */
      cio_chunk_counter_total_sub(ctx);
  }
  
  /*
   * Write at a specific offset of the content area. Offset must be >= 0 and
   * less than current data length.
   */
  int cio_chunk_write_at(struct cio_chunk *ch, off_t offset,
                         const void *buf, size_t count)
  {
      int type;
      struct cio_memfs *mf;
      struct cio_file *cf;
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          mf = ch->backend;
          mf->buf_len = offset;
      }
      else if (type == CIO_STORE_FS) {
          cf = ch->backend;
          cf->data_size = offset;
      }
  
      /*
       * By default backends (fs, mem) appends data after the it last position,
       * so we just adjust the content size to the given offset.
       */
      return cio_chunk_write(ch, buf, count);
  }
  
  int cio_chunk_write(struct cio_chunk *ch, const void *buf, size_t count)
  {
      int ret = 0;
      int type;
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          ret = cio_memfs_write(ch, buf, count);
      }
      else if (type == CIO_STORE_FS) {
          ret = cio_file_write(ch, buf, count);
      }
  
      return ret;
  }
  
  int cio_chunk_sync(struct cio_chunk *ch)
  {
      int ret = 0;
      int type;
  
      type = ch->st->type;
      if (type == CIO_STORE_FS) {
          ret = cio_file_sync(ch);
      }
  
      return ret;
  }
  
  int cio_chunk_get_content(struct cio_chunk *ch, char **buf, size_t *size)
  {
      int ret = 0;
      int type;
      struct cio_memfs *mf;
      struct cio_file *cf;
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          mf = ch->backend;
          *size = mf->buf_len;
          *buf = mf->buf_data;
          return ret;
      }
      else if (type == CIO_STORE_FS) {
          cf = ch->backend;
          ret = cio_file_read_prepare(ch->ctx, ch);
          if (ret != CIO_OK) {
              return ret;
          }
          *size = cf->data_size;
          *buf = cio_file_st_get_content(cf->map);
          return ret;
      }
  
      return CIO_ERROR;
  }
  
  /* Using the content of the chunk, generate a copy using the heap */
  int cio_chunk_get_content_copy(struct cio_chunk *ch,
                                 void **out_buf, size_t *out_size)
  {
      int type;
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          return cio_memfs_content_copy(ch, out_buf, out_size);
      }
      else if (type == CIO_STORE_FS) {
          return cio_file_content_copy(ch, out_buf, out_size);
      }
  
      return CIO_ERROR;
  }
  
  size_t cio_chunk_get_content_end_pos(struct cio_chunk *ch)
  {
      int type;
      off_t pos = 0;
      struct cio_memfs *mf;
      struct cio_file *cf;
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          mf = ch->backend;
          pos = (off_t) (mf->buf_data + mf->buf_len);
      }
      else if (type == CIO_STORE_FS) {
          cf = ch->backend;
          pos = (off_t) (cio_file_st_get_content(cf->map) + cf->data_size);
      }
  
      return pos;
  }
  
  ssize_t cio_chunk_get_content_size(struct cio_chunk *ch)
  {
      int type;
      struct cio_memfs *mf;
      struct cio_file *cf;
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          mf = ch->backend;
          return mf->buf_len;
      }
      else if (type == CIO_STORE_FS) {
          cf = ch->backend;
          return cf->data_size;
      }
  
      return -1;
  }
  
  ssize_t cio_chunk_get_real_size(struct cio_chunk *ch)
  {
      int type;
      struct cio_memfs *mf;
      struct cio_file *cf;
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          mf = ch->backend;
          return mf->buf_len;
      }
      else if (type == CIO_STORE_FS) {
          cf = ch->backend;
          return cf->fs_size;
      }
  
      return -1;
  }
  
  void cio_chunk_close_stream(struct cio_stream *st)
  {
      struct mk_list *tmp;
      struct mk_list *head;
      struct cio_chunk *ch;
  
      mk_list_foreach_safe(head, tmp, &st->chunks) {
          ch = mk_list_entry(head, struct cio_chunk, _head);
          cio_chunk_close(ch, CIO_FALSE);
      }
  }
  
  char *cio_chunk_hash(struct cio_chunk *ch)
  {
      if (ch->st->type == CIO_STORE_FS) {
          return cio_file_hash(ch->backend);
      }
  
      return NULL;
  }
  
  int cio_chunk_lock(struct cio_chunk *ch)
  {
      if (ch->lock == CIO_TRUE) {
          return CIO_ERROR;
      }
  
      ch->lock = CIO_TRUE;
  
      if (cio_chunk_is_up(ch) == CIO_TRUE) {
          return cio_chunk_sync(ch);
      }
  
      return CIO_OK;
  }
  
  int cio_chunk_unlock(struct cio_chunk *ch)
  {
      if (ch->lock == CIO_FALSE) {
          return CIO_ERROR;
      }
  
      ch->lock = CIO_FALSE;
      return CIO_OK;
  }
  
  int cio_chunk_is_locked(struct cio_chunk *ch)
  {
      return ch->lock;
  }
  
  /*
   * Start a transaction context: it keep a state of the current calculated
   * CRC32 (if enabled) and the current number of bytes in the content
   * area.
   */
  int cio_chunk_tx_begin(struct cio_chunk *ch)
  {
      int type;
      struct cio_memfs *mf;
      struct cio_file *cf;
  
      if (cio_chunk_is_locked(ch)) {
          return CIO_RETRY;
      }
  
      if (ch->tx_active == CIO_TRUE) {
          return CIO_OK;
      }
  
      ch->tx_active = CIO_TRUE;
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          mf = ch->backend;
          ch->tx_crc = mf->crc_cur;
          ch->tx_content_length = mf->buf_len;
      }
      else if (type == CIO_STORE_FS) {
          cf = ch->backend;
          ch->tx_crc = cf->crc_cur;
          ch->tx_content_length = cf->data_size;
      }
  
      return CIO_OK;
  }
  
  /*
   * Commit transaction changes, reset transaction context and leave new
   * changes in place.
   */
  int cio_chunk_tx_commit(struct cio_chunk *ch)
  {
      int ret;
  
      ret = cio_chunk_sync(ch);
      if (ret == -1) {
          return CIO_ERROR;
      }
  
      ch->tx_active = CIO_FALSE;
      return CIO_OK;
  }
  
  /*
   * Drop changes done since a transaction was initiated */
  int cio_chunk_tx_rollback(struct cio_chunk *ch)
  {
      int type;
      struct cio_memfs *mf;
      struct cio_file *cf;
  
      if (ch->tx_active == CIO_FALSE) {
          return -1;
      }
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          mf = ch->backend;
          mf->crc_cur = ch->tx_crc;
          mf->buf_len = ch->tx_content_length;
      }
      else if (type == CIO_STORE_FS) {
          cf = ch->backend;
          cf->crc_cur = ch->tx_crc;
          cf->data_size = ch->tx_content_length;
      }
  
      ch->tx_active = CIO_FALSE;
      return CIO_OK;
  }
  
  /*
   * Determinate if a Chunk content is available in memory for I/O operations. For
   * Memory backend this is always true, for Filesystem backend it checks if the
   * memory map exists and file descriptor is open.
   */
  int cio_chunk_is_up(struct cio_chunk *ch)
  {
      int type;
      struct cio_file *cf;
  
      type = ch->st->type;
      if (type == CIO_STORE_MEM) {
          return CIO_TRUE;
      }
      else if (type == CIO_STORE_FS) {
          cf = ch->backend;
          return cio_file_is_up(ch, cf);
      }
  
      return CIO_FALSE;
  }
  
  int cio_chunk_is_file(struct cio_chunk *ch)
  {
      int type;
  
      type = ch->st->type;
      if (type == CIO_STORE_FS) {
          return CIO_TRUE;
      }
  
      return CIO_FALSE;
  }
  
  static inline void chunk_state_sync(struct cio_chunk *ch)
  {
      struct cio_stream *st;
  
      if (!ch) {
          return;
      }
  
      mk_list_del(&ch->_state_head);
      st = ch->st;
      if (cio_chunk_is_up(ch) == CIO_TRUE) {
          mk_list_add(&ch->_state_head, &st->chunks_up);
      }
      else {
          mk_list_add(&ch->_state_head, &st->chunks_down);
      }
  }
  
  int cio_chunk_down(struct cio_chunk *ch)
  {
      int ret;
      int type;
  
      type = ch->st->type;
      if (type == CIO_STORE_FS) {
          ret = cio_file_down(ch);
          chunk_state_sync(ch);
          return ret;
      }
  
      return CIO_OK;
  }
  
  int cio_chunk_up(struct cio_chunk *ch)
  {
      int ret;
      int type;
  
      type = ch->st->type;
      if (type == CIO_STORE_FS) {
          ret = cio_file_up(ch);
          chunk_state_sync(ch);
          return ret;
      }
  
      return CIO_OK;
  }
  
  int cio_chunk_up_force(struct cio_chunk *ch)
  {
      int ret;
      int type;
  
      type = ch->st->type;
      if (type == CIO_STORE_FS) {
          ret = cio_file_up_force(ch);
          chunk_state_sync(ch);
          return ret;
      }
  
      return CIO_OK;
  }
  
  char *cio_version()
  {
      return CIO_VERSION_STR;
  }
  
  /*
   * Counters API
   */
  
  /* Increase the number of total chunks registered (+1) */
  size_t cio_chunk_counter_total_add(struct cio_ctx *ctx)
  {
      ctx->total_chunks++;
      return ctx->total_chunks;
  }
  
  /* Decrease the total number of chunks (-1) */
  size_t cio_chunk_counter_total_sub(struct cio_ctx *ctx)
  {
      ctx->total_chunks--;
      return ctx->total_chunks;
  }
  
  /* Increase the number of total chunks up in memory (+1) */
  size_t cio_chunk_counter_total_up_add(struct cio_ctx *ctx)
  {
      ctx->total_chunks_up++;
      return ctx->total_chunks_up;
  }
  
  /* Decrease the total number of chunks up in memory (-1) */
  size_t cio_chunk_counter_total_up_sub(struct cio_ctx *ctx)
  {
      ctx->total_chunks_up--;
      return ctx->total_chunks_up;
  }
  