/******************************************************
Copyright (c) 2012 Percona LLC and/or its affiliates.

tmpfile datasink for XtraBackup.

This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; version 2 of the License.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA

*******************************************************/

/* Do all writes to temporary files first, then pipe them to the specified
datasink in a serialized way in deinit(). */

#include <my_global.h>
#include <my_base.h>
#include "common.h"
#include "datasink.h"

typedef struct {
 pthread_mutex_t  mutex;
 LIST  *file_list;
} ds_tmpfile_ctxt_t;

typedef struct {
 LIST   list;
 File   fd;
 char  *orig_path;
 MY_STAT   mystat;
 ds_file_t *file;
} ds_tmp_file_t;

static ds_ctxt_t *tmpfile_init(const char *root);
static ds_file_t *tmpfile_open(ds_ctxt_t *ctxt, const char *path,
   const MY_STAT *mystat, bool rewrite);
static int tmpfile_write(ds_file_t *file, const uchar *buf, size_t len);
static int tmpfile_close(ds_file_t *file);
static void tmpfile_deinit(ds_ctxt_t *ctxt);

datasink_t datasink_tmpfile = {
 &tmpfile_init,
 &tmpfile_open,
 &tmpfile_write,
 nullptr,
 &tmpfile_close,
 &dummy_remove,
 nullptr,
 nullptr,
 &tmpfile_deinit
};


static ds_ctxt_t *
tmpfile_init(const char *root)
{
 ds_ctxt_t  *ctxt;
 ds_tmpfile_ctxt_t *tmpfile_ctxt;

 ctxt = (ds_ctxt_t *)my_malloc(PSI_NOT_INSTRUMENTED,
                   sizeof(ds_ctxt_t) + sizeof(ds_tmpfile_ctxt_t), MYF(MY_FAE));
 tmpfile_ctxt = (ds_tmpfile_ctxt_t *) (ctxt + 1);
 tmpfile_ctxt->file_list = NULL;
 if (pthread_mutex_init(&tmpfile_ctxt->mutex, NULL)) {

  my_free(ctxt);
  return NULL;
 }

 ctxt->ptr = tmpfile_ctxt;
 ctxt->root = my_strdup(PSI_NOT_INSTRUMENTED, root, MYF(MY_FAE));

 return ctxt;
}

static ds_file_t *
tmpfile_open(ds_ctxt_t *ctxt, const char *path,
   const MY_STAT *mystat, bool rewrite)
{
  DBUG_ASSERT(rewrite == false);
 ds_tmpfile_ctxt_t *tmpfile_ctxt;
 char    tmp_path[FN_REFLEN];
 ds_tmp_file_t  *tmp_file;
 ds_file_t  *file;
 size_t    path_len;
 File    fd;

 /* Create a temporary file in tmpdir. The file will be automatically
removed on close. Code copied from mysql_tmpfile(). */
 fd = create_temp_file(tmp_path,xtrabackup_tmpdir,
         "xbtemp", O_BINARY | O_SEQUENTIAL,
         MYF(MY_WME | MY_TEMPORARY));

 if (fd < 0) {
  return NULL;
 }

 path_len = strlen(path) + 1; /* terminating '\0' */

 file = (ds_file_t *) my_malloc(PSI_NOT_INSTRUMENTED,
                        sizeof(ds_file_t) + sizeof(ds_tmp_file_t) + path_len, MYF(MY_FAE));

 tmp_file = (ds_tmp_file_t *) (file + 1);
 tmp_file->file = file;
 memcpy(&tmp_file->mystat, mystat, sizeof(MY_STAT));
 /* Save a copy of 'path', since it may not be accessible later */
 tmp_file->orig_path = (char *) tmp_file + sizeof(ds_tmp_file_t);

 tmp_file->fd = fd;
 posix_fadvise(tmp_file->fd, 0, 0, POSIX_FADV_DONTNEED);
 memcpy(tmp_file->orig_path, path, path_len);

 /* Store the real temporary file name in file->path */
 file->path = my_strdup(PSI_NOT_INSTRUMENTED, tmp_path, MYF(MY_FAE));
 file->ptr = tmp_file;

 /* Store the file object in the list to be piped later */
 tmpfile_ctxt = (ds_tmpfile_ctxt_t *) ctxt->ptr;
 tmp_file->list.data = tmp_file;

 pthread_mutex_lock(&tmpfile_ctxt->mutex);
 tmpfile_ctxt->file_list = list_add(tmpfile_ctxt->file_list,
        &tmp_file->list);
 pthread_mutex_unlock(&tmpfile_ctxt->mutex);

 return file;
}

static int
tmpfile_write(ds_file_t *file, const uchar *buf, size_t len)
{
 File fd = ((ds_tmp_file_t *) file->ptr)->fd;

 if (my_write(fd, buf, len, MYF(MY_WME | MY_NABP))) {
  return 1;
 }

 return 0;
}

static int
tmpfile_close(ds_file_t *file)
{
 /* Do nothing -- we will close (and thus remove) the file after piping
it to the destination datasink in tmpfile_deinit(). */

 my_free(file->path);

 return 0;
}

static void
tmpfile_deinit(ds_ctxt_t *ctxt)
{
 LIST   *list;
 ds_tmpfile_ctxt_t *tmpfile_ctxt;
 MY_STAT    mystat;
 ds_tmp_file_t   *tmp_file;
 ds_file_t  *dst_file;
 ds_ctxt_t  *pipe_ctxt;
 void   *buf = NULL;
 const size_t   buf_size = 10 * 1024 * 1024;
 size_t    bytes;
 size_t    offset;

 pipe_ctxt = ctxt->pipe_ctxt;
 xb_a(pipe_ctxt != NULL);

 buf = my_malloc(PSI_NOT_INSTRUMENTED, buf_size, MYF(MY_FAE));

 tmpfile_ctxt = (ds_tmpfile_ctxt_t *) ctxt->ptr;
 list = tmpfile_ctxt->file_list;

 /* Walk the files in the order they have been added */
 list = list_reverse(list);
 while (list != NULL) {
  tmp_file = (ds_tmp_file_t *)list->data;
  /* Stat the file to replace size and mtime on the original
* mystat struct */
  if (my_fstat(tmp_file->fd, &mystat, MYF(0))) {
   die("my_fstat() failed.");
  }
  tmp_file->mystat.st_size = mystat.st_size;
  tmp_file->mystat.st_mtime = mystat.st_mtime;

  dst_file = ds_open(pipe_ctxt, tmp_file->orig_path,
       &tmp_file->mystat);
  if (dst_file == NULL) {
   die("could not stream a temporary file to "
       "'%s'", tmp_file->orig_path);
  }

  /* copy to the destination datasink */
  posix_fadvise(tmp_file->fd, 0, 0, POSIX_FADV_SEQUENTIAL);
  if (my_seek(tmp_file->fd, 0, SEEK_SET, MYF(0)) ==
      MY_FILEPOS_ERROR) {
   die("my_seek() failed for '%s', errno = %d.",
       tmp_file->file->path, my_errno);
  }
  offset = 0;
  while ((bytes = my_read(tmp_file->fd, (unsigned char *)buf, buf_size,
     MYF(MY_WME))) > 0) {
   posix_fadvise(tmp_file->fd, offset, buf_size, POSIX_FADV_DONTNEED);
   offset += buf_size;
   if (ds_write(dst_file, buf, bytes)) {
    die("cannot write to stream for '%s'.",
        tmp_file->orig_path);
   }
  }
  if (bytes == (size_t) -1) {
   die("my_read failed for %s", tmp_file->orig_path);
  }

  my_close(tmp_file->fd, MYF(MY_WME));
  ds_close(dst_file);

  list = list_rest(list);
  my_free(tmp_file->file);
 }

 pthread_mutex_destroy(&tmpfile_ctxt->mutex);

 my_free(buf);
 my_free(ctxt->root);
 my_free(ctxt);
}