text: /**
   * @file   video_capture/import.cpp
   * @author Martin Pulec     <pulec@cesnet.cz>
   */
  /*
   * Copyright (c) 2012-2019 CESNET, z. s. p. o.
   * All rights reserved.
   *
   * Redistribution and use in source and binary forms, with or without
   * modification, is permitted provided that the following conditions
   * are met:
   *
   * 1. Redistributions of source code must retain the above copyright
   *    notice, this list of conditions and the following disclaimer.
   *
   * 2. Redistributions in binary form must reproduce the above copyright
   *    notice, this list of conditions and the following disclaimer in the
   *    documentation and/or other materials provided with the distribution.
   *
   * 3. Neither the name of CESNET nor the names of its contributors may be
   *    used to endorse or promote products derived from this software without
   *    specific prior written permission.
   *
   * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS
   * "AS IS" AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING,
   * BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
   * AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
   * EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
   * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
   * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
   * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
   * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
   * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
   * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
   */
  
  #ifdef HAVE_CONFIG_H
  #include "config.h"
  #include "config_unix.h"
  #include "config_win32.h"
  #endif // HAVE_CONFIG_H
  
  #include "debug.h"
  #include "host.h"
  #include "lib_common.h"
  #include "video.h"
  #include "video_capture.h"
  
  #include "tv.h"
  
  #include "audio/types.h"
  #include "audio/wav_reader.h"
  #include "keyboard_control.h"
  #include "messaging.h"
  #include "module.h"
  #include "playback.h"
  #include "utils/ring_buffer.h"
  #include "utils/worker.h"
  #include "video_export.h"
  
  #include <stdio.h>
  #include <stdlib.h>
  #include <fcntl.h>
  #include <sstream>
  #include <string>
  #include <string.h>
  #include <sys/stat.h>
  #include <sys/time.h>
  #include <sys/types.h>
  #include <unistd.h>
  
  #include <condition_variable>
  #include <chrono>
  #include <list>
  #include <mutex>
  #include <utility> // pair
  
  #define BUFFER_LEN_MAX 40
  #define MAX_CLIENTS 16
  
  #define VIDCAP_IMPORT_ID 0x76FA7F6D
  
  #define PIPE "/tmp/ultragrid_import.fifo"
  
  #define MAX_NUMBER_WORKERS 100
  #define MOD_NAME "[import] "
  
  using std::condition_variable;
  using std::chrono::duration;
  using std::list;
  using std::min;
  using std::max;
  using std::mutex;
  using std::ostringstream;
  using std::pair;
  using std::string;
  using std::unique_lock;
  
  struct processed_entry;
  struct tile_data {
          char *data;
          int data_len;
  };
  
  struct processed_entry {
          struct processed_entry *next;
          int count;
          struct tile_data tiles[];
  };
  
  typedef enum {
          SEEK,
          FINALIZE,
          PAUSE
  } import_message_t;
  
  struct import_message;
  
  struct import_message {
          import_message_t       type;
          void           *data;
          size_t          data_len;
          struct import_message *next;
  };
  
  typedef enum {
          IMPORT_SEEK_SET,
          IMPORT_SEEK_END,
          IMPORT_SEEK_CUR
  } seek_direction_t;
  
  struct seek_data {
          seek_direction_t whence;
          ssize_t offset;
  };
  
  struct message_queue {
          struct import_message *head;
          struct import_message *tail;
          size_t          len;
  };
  
  struct audio_state {
          bool has_audio;
          FILE *file;
          struct wav_metadata metadata;
          ring_buffer_t *data;
          int total_samples;
          int samples_read;
          pthread_t thread_id;
  
          condition_variable worker_cv;
          condition_variable boss_cv;
  
          mutex lock;
          unsigned long long int played_samples;
  
          struct message_queue message_queue;
  }; 
  
  struct vidcap_import_state {
          struct module mod;
          struct module *parent;
          struct audio_frame audio_frame;
          struct audio_state audio_state;
          struct video_desc video_desc;
          int frames;
          int frames_prev;
          struct timeval t0;
          char *directory;
          char tile_delim; // eg. '_' for format "00000001_0.yuv"
  
          struct message_queue message_queue;
  
          mutex lock;
          condition_variable worker_cv;
          condition_variable boss_cv;
          struct processed_entry * volatile head, * volatile tail;
          volatile int queue_len;
  
          pthread_t thread_id;
  
          struct timeval prev_time;
          int count;
  
          bool finished;
          bool loop;
          bool o_direct;
          int video_reading_threads_count;
          bool should_exit_at_end;
          double force_fps;
  };
  
  static void * audio_reading_thread(void *args);
  static void * reading_thread(void *args);
  static bool init_audio(struct vidcap_import_state *state, char *audio_filename);
  static void send_message(struct import_message *msg, struct message_queue *queue);
  static struct import_message* pop_message(struct message_queue *queue);
  static int flush_processed(struct processed_entry *list);
  static void message_queue_clear(struct message_queue *queue);
  static void vidcap_import_new_message(struct module *);
  static void process_msg(struct vidcap_import_state *state, const char *message);
  
  static void cleanup_common(struct vidcap_import_state *s);
  
  static void message_queue_clear(struct message_queue *queue) {
          queue->head = queue->tail = NULL;
          queue->len = 0;
  }
  
  static struct vidcap_type *
  vidcap_import_probe(bool verbose, void (**deleter)(void *))
  {
          UNUSED(verbose);
          *deleter = free;
  	struct vidcap_type*		vt;
      
  	vt = (struct vidcap_type *) calloc(1, sizeof(struct vidcap_type));
  	if (vt != NULL) {
  		vt->name        = "import";
  		vt->description = "Video importer (not to be called directly)";
  	}
  	return vt;
  }
  
  #define READ_N(buf, len) if (fread(buf, len, 1, audio_file) != 1) goto error_format;
  
  static bool init_audio(struct vidcap_import_state *s, char *audio_filename)
  {
          FILE *audio_file = fopen(audio_filename, "rb");
          if(!audio_file) {
                  perror("Cannot open audio file");
                  return false;
          }
  
          int ret = read_wav_header(audio_file, &s->audio_state.metadata);
          if (ret != WAV_HDR_PARSE_OK) {
                          log_msg(LOG_LEVEL_ERROR, "%s!\n", get_wav_error(ret));
                          goto error_format;
          }
  
          s->audio_frame.ch_count = s->audio_state.metadata.ch_count;
          s->audio_frame.sample_rate = s->audio_state.metadata.sample_rate;
          s->audio_frame.bps = s->audio_state.metadata.bits_per_sample / 8;
          s->audio_state.total_samples = s->audio_state.metadata.data_size / s->audio_frame.bps / s->audio_frame.ch_count;
          s->audio_state.samples_read = 0;
  
          s->audio_state.data = ring_buffer_init(s->audio_frame.bps * s->audio_frame.sample_rate *
                          s->audio_frame.ch_count * 180);
  
          s->audio_frame.max_size = s->audio_frame.bps * s->audio_frame.sample_rate * s->audio_frame.ch_count;
          s->audio_frame.data_len = 0;
          s->audio_frame.data = (char *) malloc(s->audio_frame.max_size);
  
          s->audio_state.file = audio_file;
  
          s->audio_state.played_samples = 0;
  
          return true;
  
  error_format:
          fprintf(stderr, "Audio format file error - unknown format\n");
          fclose(audio_file);
          return false;
  }
  
  static int
  vidcap_import_init(struct vidcap_params *params, void **state)
  {
  	struct vidcap_import_state *s = NULL;
          FILE *info = NULL; // metadata file
          char *tmp = strdup(vidcap_params_get_fmt(params));
          bool disable_audio = false;
  
  try {
  	printf("vidcap_import_init\n");
  
          if (strlen(tmp) == 0 || strcmp(tmp, "help") == 0) {
                  printf("Import usage:\n"
                                  "\t<directory>{:loop|:mt_reading=<nr_threads>|:o_direct|:exit_at_end|:fps=<fps>|:disable_audio}\n"
                                  "\t\t<fps> - overrides FPS from sequence metadata\n");
                  delete s;
                  free(tmp);
                  return VIDCAP_INIT_NOERR;
          }
  
          s = new vidcap_import_state();
          s->head = s->tail = NULL;
          s->queue_len = 0;
          s->frames_prev = s->frames = 0;
          gettimeofday(&s->t0, NULL);
  
          s->parent = vidcap_params_get_parent(params);
          module_init_default(&s->mod);
          s->mod.cls = MODULE_CLASS_DATA;
          s->mod.priv_data = s;
          s->mod.new_message = vidcap_import_new_message;
          module_register(&s->mod, s->parent);
  
          s->video_reading_threads_count = 1; // default is single threaded
  
          char *save_ptr = NULL;
          char *suffix;
          s->directory = strtok_r(tmp, ":", &save_ptr);
          if (s->directory == nullptr) {
                  throw string("Wrong directory name!\n");
          }
          s->directory = strdup(s->directory); // make a copy
  
          while ((suffix = strtok_r(NULL, ":", &save_ptr)) != NULL) {
                  if (suffix[0] == '\\') { // MSW path
                          assert(strlen(s->directory) == 1); // c:\something -> should be 'c'
                          char *tmp = (char *) malloc(2 + strlen(suffix) + 1);
                          sprintf(tmp, "%c:%s", s->directory[0], suffix);
                          free(s->directory);
                          s->directory = tmp;
                  } else if (strcmp(suffix, "loop") == 0) {
                          s->loop = true;
                  } else if (strncmp(suffix, "mt_reading=",
                                          strlen("mt_reading=")) == 0) {
                          s->video_reading_threads_count = atoi(suffix +
                                          strlen("mt_reading="));
                          assert(s->video_reading_threads_count <=
                                          MAX_NUMBER_WORKERS);
                  } else if (strcmp(suffix, "o_direct") == 0) {
                          s->o_direct = true;
                  } else if (strcmp(suffix, "noaudio") == 0) {
                          disable_audio = true;
                  } else if (strcmp(suffix, "opportunistic_audio") == 0) { // skip
                  } else if (strcmp(suffix, "exit_at_end") == 0) {
                          s->should_exit_at_end = true;
                  } else if (strncmp(suffix, "fps=", strlen("fps=")) == 0) {
                          s->force_fps = atof(suffix + strlen("fps="));
                  } else {
                          throw string("[Playback] Unrecognized"
                                          " option ") + suffix + ".\n";
                  }
          }
          free(tmp);
          tmp = NULL;
  
          // strip video.info if user included in path
          if (strstr(s->directory, "video.info") == s->directory) {
                  strcpy(s->directory, ".");
          }
          if (strstr(s->directory, "/video.info") != nullptr) {
                  *strrchr(s->directory, '/') = '\0';
          }
  
          message_queue_clear(&s->message_queue);
          message_queue_clear(&s->audio_state.message_queue);
  
          char *audio_filename = (char *) malloc(strlen(s->directory) + strlen("/sound.wav") + 1);
          assert(audio_filename != NULL);
          strcpy(audio_filename, s->directory);
          strcat(audio_filename, "/sound.wav");
          if((vidcap_params_get_flags(params) & VIDCAP_FLAG_AUDIO_EMBEDDED) && !disable_audio && init_audio(s, audio_filename)) {
                  s->audio_state.has_audio = true;
                  if(pthread_create(&s->audio_state.thread_id, NULL, audio_reading_thread, (void *) s) != 0) {
                          free(audio_filename);
                          throw string("Unable to create thread.\n");
                  }
          } else {
                  s->audio_state.has_audio = false;
          }
          free(audio_filename);
          
          char *info_filename = (char *) malloc(strlen(s->directory) + sizeof("/video.info") + 1);
          assert(info_filename != NULL);
          strcpy(info_filename, s->directory);
          strcat(info_filename, "/video.info");
  
          info = fopen(info_filename, "r");
          free(info_filename);
          if(info == NULL) {
                  perror("[import] Failed to open index file");
                  if (errno == ENOENT) {
                          throw string("Invalid directory?\n");
                  }
                  throw string();
          }
  
          struct video_desc desc;
          memset(&desc, 0, sizeof desc);
  
          char line[512];
          uint32_t items_found = 0;
          while(!feof(info)) {
                  if(fgets(line, sizeof(line), info) == NULL) {
                          // empty line
                          continue;
                  }
                  if(strncmp(line, "version ", strlen("version ")) == 0) {
                          long int version = strtol(line + strlen("version "), (char **) NULL, 10);
                          if(version == LONG_MIN || version == LONG_MAX) {
                                  throw string("[import] cannot read version line.\n");
                          }
                          if(version != VIDEO_EXPORT_SUMMARY_VERSION) {
                                  ostringstream oss;
                                  oss << "[import] Invalid version " << version << ".\n";
                                  throw oss.str();
                          }
                          items_found |= 1<<0;
                  } else if(strncmp(line, "width ", strlen("width ")) == 0) {
                          long int width = strtol(line + strlen("width "), (char **) NULL, 10);
                          if(width == LONG_MIN || width == LONG_MAX) {
                                  throw string("[import] cannot read video width.\n");
                          }
                          desc.width = width;
                          items_found |= 1<<1;
                  } else if(strncmp(line, "height ", strlen("height ")) == 0) {
                          long int height = strtol(line + strlen("height "), (char **) NULL, 10);
                          if(height == LONG_MIN || height == LONG_MAX) {
                                  throw string("[import] cannot read video height.\n");
                          }
                          desc.height = height;
                          items_found |= 1<<2;
                  } else if(strncmp(line, "fourcc ", strlen("fourcc ")) == 0) {
                          char *ptr = line + strlen("fourcc ");
                          if(strlen(ptr) != 5) { // including '\n'
                                  throw string("[import] cannot read video FourCC tag.\n");
                          }
                          uint32_t fourcc;
                          memcpy((void *) &fourcc, ptr, sizeof(fourcc));
                          desc.color_spec = get_codec_from_fcc(fourcc);
                          if(desc.color_spec == VIDEO_CODEC_NONE) {
                                  throw string("[import] Requested codec not known.\n");
                          }
                          items_found |= 1<<3;
                  } else if(strncmp(line, "fps ", strlen("fps ")) == 0) {
                          char *ptr = line + strlen("fps ");
                          desc.fps = strtod(ptr, NULL);
                          if(desc.fps == HUGE_VAL || desc.fps <= 0) {
                                  throw string("[import] Invalid FPS.\n");
                          }
                          items_found |= 1<<4;
                  } else if(strncmp(line, "interlacing ", strlen("interlacing ")) == 0) {
                          char *ptr = line + strlen("interlacing ");
                          desc.interlacing = (interlacing_t) atoi(ptr);
                          if(desc.interlacing > 4) {
                                  throw string("[import] Invalid interlacing.\n");
                          }
                          items_found |= 1<<5;
                  } else if(strncmp(line, "count ", strlen("count ")) == 0) {
                          char *ptr = line + strlen("count ");
                          s->count = atoi(ptr);
                          items_found |= 1<<6;
                  }
          }
  
          // override metadata fps setting
          if (s->force_fps > 0.0) {
                  desc.fps = s->force_fps;
          }
  
          assert(desc.color_spec != VIDEO_CODEC_NONE && desc.width != 0 && desc.height != 0 && desc.fps != 0.0 &&
                          s->count != 0);
  
          char name[1024];
          snprintf(name, sizeof(name), "%s/%08d.%s", s->directory, 1,
                          get_codec_file_extension(desc.color_spec));
  
          struct stat sb;
          if (stat(name, &sb) == 0) {
                  desc.tile_count = 1;
          } else {
                  desc.tile_count = 0;
                  char possible_tile_delim[] = { '_', '-' };
                  for (unsigned int d = 0; d < sizeof possible_tile_delim; d++) {
                          for (int i = 0; i < 10; i++) {
                                  snprintf(name, sizeof(name), "%s/%08d%c%d.%s",
                                                  s->directory, 1,
                                                  possible_tile_delim[d], i,
                                                  get_codec_file_extension(desc.color_spec));
                                  if (stat(name, &sb) == 0) {
                                          desc.tile_count++;
                                  } else {
                                          break;
                                  }
                          }
                          if (desc.tile_count > 0) {
                                  s->tile_delim = possible_tile_delim[d];
                                  break;
                          }
                  }
                  if (desc.tile_count == 0) {
                          throw string("Unable to open first file of "
                                          "the video sequence.\n");
                  }
          }
  
          s->video_desc = desc;
  
          fclose(info);
          info = NULL;
  
          if(items_found != (1 << 7) - 1) {
                  throw string("[import] Failed while reading config file - some items missing.\n");
          }
  
          if(pthread_create(&s->thread_id, NULL, reading_thread, (void *) s) != 0) {
                  throw string("Unable to create thread.\n");
          }
  
          gettimeofday(&s->prev_time, NULL);
  
          playback_register_keyboard_ctl(&s->mod);
  
          *state = s;
  	return VIDCAP_INIT_OK;
  } catch (string const & str) {
          LOG(LOG_LEVEL_ERROR) << MOD_NAME << str;
          free(tmp);
          if (info != NULL)
                  fclose(info);
          cleanup_common(s);
          delete s;
          return VIDCAP_INIT_FAIL;
  }
  }
  
  static void exit_reading_threads(struct vidcap_import_state *s)
  {
          struct import_message *msg = (struct import_message *) malloc(sizeof(struct import_message));
  
          msg->type = FINALIZE;
          msg->data = NULL;
          msg->data_len = 0;
          msg->next = NULL;
  
          {
                  unique_lock<mutex> lk(s->lock);
                  send_message(msg, &s->message_queue);
                  lk.unlock();
                  s->worker_cv.notify_one();
          }
  
  	pthread_join(s->thread_id, NULL);
  
          // audio
          if(s->audio_state.has_audio) {
                  struct import_message *msg = (struct import_message *) malloc(sizeof(struct import_message));
  
                  msg->type = FINALIZE;
                  msg->data = NULL;
                  msg->data_len = 0;
                  msg->next = NULL;
  
                  {
                          unique_lock<mutex> lk(s->audio_state.lock);
                          send_message(msg, &s->audio_state.message_queue);
                          lk.unlock();
                          s->audio_state.worker_cv.notify_one();
                  }
  
                  pthread_join(s->audio_state.thread_id, NULL);
          }
  }
  
  static void free_entry(struct processed_entry *entry)
  {
          if (entry == NULL) {
                  return;
          }
          for (int i = 0; i < entry->count; ++i) {
                  aligned_free(entry->tiles[i].data);
          }
  
          free(entry);
  }
  
  static void vidcap_import_finish(void *state)
  {
          struct vidcap_import_state *s = (struct vidcap_import_state *) state;
  
          exit_reading_threads(s);
  }
  
  static int flush_processed(struct processed_entry *list)
  {
          int frames_deleted = 0;
          struct processed_entry *current = list;
  
          while(current != NULL) {
                  struct processed_entry *tmp = current;
                  current = current->next;
                  free_entry(tmp);
                  frames_deleted++;
          }
  
          return frames_deleted;
  }
  
  static void cleanup_common(struct vidcap_import_state *s) {
          flush_processed(s->head);
  
          free(s->directory);
  
          // audio
          if(s->audio_state.has_audio) {
                  ring_buffer_destroy(s->audio_state.data);
  
                  free(s->audio_frame.data);
  
                  fclose(s->audio_state.file);
          }
  
          module_done(&s->mod);
  }
  
  static void vidcap_import_done(void *state)
  {
  	struct vidcap_import_state *s = (struct vidcap_import_state *) state;
  	assert(s != NULL);
  
          vidcap_import_finish(state);
  
          cleanup_common(s);
          delete s;
  }
  
  static void send_message(struct import_message *msg, struct message_queue *queue)
  {
          if(queue->head) {
                  queue->tail->next = msg;
                  queue->tail = msg;
          } else {
                  queue->head = queue->tail = msg;
          }
  
          queue->len += 1;
  }
  
  static struct import_message *pop_message(struct message_queue *queue)
  {
          assert(queue->len > 0);
          struct import_message *ret;
  
          ret = queue->head;
          queue->head = queue->head->next;
          if(queue->head == NULL) {
                  queue->tail = NULL;
          }
  
          queue->len -= 1;
  
          return ret;
  }
  
  static void vidcap_import_new_message(struct module *mod) {
          struct msg_universal *m;
          while ((m = (struct msg_universal *) check_message(mod))) {
                  process_msg((vidcap_import_state *) mod->priv_data, m->text);
                  free_message((struct message *) m, new_response(RESPONSE_ACCEPTED, "import is processing the request"));
          }
  }
  
  static void process_msg(struct vidcap_import_state *s, const char *message)
  {
          if(strcasecmp(message, "pause") == 0) {
                  struct import_message *msg = (struct import_message *) malloc(sizeof(struct import_message));
                  msg->type = PAUSE;
                  msg->data = NULL;
                  msg->data_len = 0;
                  msg->next = NULL;
  
                  unique_lock<mutex> lk(s->lock);
                  send_message(msg, &s->message_queue);
                  lk.unlock();
                  s->worker_cv.notify_one();
          } else if(strncasecmp(message, "seek ", strlen("seek ")) == 0) {
                  if(s->audio_state.has_audio == true) {
                          fprintf(stderr, "Seeking now allowed if we have audio. (Not yet implemented)\n");
                          return;
                  }
  
                  const char *time_spec = message + strlen("seek ");
  
                  struct import_message *msg = (struct import_message *) malloc(sizeof(struct import_message));
                  struct seek_data *data = (struct seek_data *) malloc(sizeof(struct seek_data));
                  msg->type = SEEK;
                  msg->data = data;
                  msg->data_len = sizeof(struct seek_data);
                  msg->next = NULL;
  
                  if(time_spec[0] == '+' || time_spec[0] == '-') {
                          data->whence = IMPORT_SEEK_CUR;
                          if(strchr(time_spec, 's') != NULL) {
                                  double val = atof(time_spec);
                                  data->offset = val * s->video_desc.fps;
                          } else {
                                  data->offset = atoi(time_spec);
                          }
                  } else {
                          data->whence = IMPORT_SEEK_SET;
                          if(strchr(time_spec, 's') != NULL) {
                                  double val = atof(time_spec);
                                  data->offset = val * s->video_desc.fps;
                          } else {
                                  data->offset = atoi(time_spec);
                          }
                  }
  
                  struct import_message *audio_msg = NULL;
  
                  if(s->audio_state.has_audio) {
                          audio_msg = (struct import_message *) malloc(sizeof(struct import_message));
                          memcpy(audio_msg, msg, sizeof(struct import_message));
  
                          if(audio_msg->data) { // deep copy
                                  audio_msg->data = malloc(msg->data_len);
                                  memcpy(audio_msg->data, msg->data, msg->data_len);
                          }
                  }
  
                  {
                          unique_lock<mutex> lk(s->lock);
                          send_message(msg, &s->message_queue);
                          lk.unlock();
                          s->worker_cv.notify_one();
                  }
  
                  if (s->audio_state.has_audio) {
                          unique_lock<mutex> lk(s->audio_state.lock);
                          send_message(audio_msg, &s->audio_state.message_queue);
                          lk.unlock();
                          s->audio_state.worker_cv.notify_one();
                  }
          } else if(strcasecmp(message, "quit") == 0) {
                  exit_uv(0);
          } else {
                  fprintf(stderr, "Warning: unknown message: \'%s\'\n", message);
          }
  }
  
  struct client;
  
  struct client {
          int fd;
          char buff[1024];
          int buff_len;
          bool pipe;
  
          struct client *next;
  };
          
  static void * audio_reading_thread(void *args)
  {
  	struct vidcap_import_state 	*s = (struct vidcap_import_state *) args;
  
          //while(s->audio_state.samples_read < s->audio_state.total_samples && !s->finish_threads) {
          while(1) {
                  int max_read;
                  {
                          unique_lock<mutex> lk(s->audio_state.lock);
                          while((ring_get_current_size(s->audio_state.data) > ring_get_size(s->audio_state.data) * 2 / 3 ||
                                                  s->audio_state.samples_read >= s->audio_state.total_samples)
                                          && s->audio_state.message_queue.len == 0) {
                                  s->audio_state.worker_cv.wait(lk);
                          }
  
                          while(s->audio_state.message_queue.len > 0) {
                                  struct import_message *msg = pop_message(&s->audio_state.message_queue);
                                  if(msg->type == FINALIZE) {
                                          free(msg);
                                          return NULL;
                                  } else if (msg->type == SEEK) {
                                          fprintf(stderr, "Seeking in audio import not yet implemented.\n");
                                          abort();
                                  }
                          }
  
                          max_read = ring_get_size(s->audio_state.data) -
                                          ring_get_current_size(s->audio_state.data) - 1;
  
                  }
  
                  char *buffer = (char *) malloc(max_read);
  
                  size_t samples = wav_read(buffer, max_read / s->audio_frame.ch_count / s->audio_frame.bps, s->audio_state.file, &s->audio_state.metadata);
                  s->audio_state.samples_read += samples;
  
                  {
                          unique_lock<mutex> lk(s->audio_state.lock);
                          ring_buffer_write(s->audio_state.data, buffer, samples * s->audio_frame.ch_count * s->audio_frame.bps);
                          lk.unlock();
                          s->audio_state.boss_cv.notify_one();
                  }
  
                  free(buffer);
          }
  
          return NULL;
  }
  
  struct video_reader_data {
          char file_name_prefix[512];
          char file_name_suffix[512];
          char tile_delim;
          unsigned int tile_count;
          struct processed_entry *entry;
          bool o_direct;
  };
  
  #define ALLOC_ALIGN 512
  
  static void *video_reader_callback(void *arg)
  {
          struct video_reader_data *data =
                  (struct video_reader_data *) arg;
         
          data->entry = (struct processed_entry *) calloc(1, sizeof(struct processed_entry) + data->tile_count * sizeof(struct tile_data));
          assert(data->entry != NULL);
          data->entry->next = NULL;
          data->entry->count = data->tile_count;
  
          for (unsigned int i = 0; i < data->tile_count; i++) {
                  char name[1048];
                  char tile_idx[3] = "";
                  if (data->tile_count > 1) {
                          sprintf(tile_idx, "%c%d", data->tile_delim, i);
  	        }
                  snprintf(name, sizeof(name), "%s%s.%s",
                                  data->file_name_prefix, tile_idx,
                                  data->file_name_suffix);
  
                  struct stat sb;
  
                  int flags = O_RDONLY;
  #ifdef WIN32
                  flags |= O_BINARY;
  #endif
                  if (data->o_direct) {
  #ifdef HAVE_LINUX
                          flags |= O_DIRECT;
  #endif
                  }
                  int fd = open(name, flags);
                  if(fd == -1) {
                          perror("open");
                          return NULL;
                  }
                  if (fstat(fd, &sb)) {
                          perror("fstat");
                          close(fd);
                          free_entry(data->entry);
                          return NULL;
                  }
  
                  data->entry->tiles[i].data_len = sb.st_size;
                  const int aligned_data_len = (data->entry->tiles[i].data_len + ALLOC_ALIGN - 1)
                          / ALLOC_ALIGN * ALLOC_ALIGN;
                  // alignment needed when using O_DIRECT flag
                  data->entry->tiles[i].data = (char *)
                          aligned_malloc(aligned_data_len, ALLOC_ALIGN);
                  assert(data->entry->tiles[i].data != NULL);
  
                  ssize_t bytes = 0;
                  do {
                          ssize_t res = read(fd, data->entry->tiles[i].data + bytes,
                                          (data->entry->tiles[i].data_len - bytes + ALLOC_ALIGN - 1)
                                          / ALLOC_ALIGN * ALLOC_ALIGN);
                          if (res <= 0) {
                                  perror("read");
                                  for (unsigned int i = 0; i < data->tile_count; i++) {
                                          aligned_free(data->entry->tiles[i].data);
                                  }
                                  free(data->entry);
                                  close(fd);
                                  return NULL;
                          }
                          bytes += res;
                  } while (bytes < data->entry->tiles[i].data_len);
  
                  close(fd);
          }
  
          return data;
  }
  
  static void * reading_thread(void *args)
  {
  	struct vidcap_import_state 	*s = (struct vidcap_import_state *) args;
          int index = 0;
  
          bool paused = false;
  
          ///while(index < s->count && !s->finish_threads) {
          while(1) {
                  {
                          unique_lock<mutex> lk(s->lock);
                          while((s->queue_len >= BUFFER_LEN_MAX - 1 || index >= s->count || paused)
                                         && s->message_queue.len == 0) {
                                  if (index >= s->count) {
                                          s->finished = true;
                                  }
                                  s->worker_cv.wait(lk);
                          }
  
                          while(s->message_queue.len > 0) {
                                  struct import_message *msg = pop_message(&s->message_queue);
                                  if(msg->type == FINALIZE) {
                                          free(msg);
                                          return NULL;
                                  } else if(msg->type == PAUSE) {
                                          paused = !paused;
                                          printf("Toggle pause\n");
  
                                          index -= flush_processed(s->head);
                                          s->queue_len = 0;
                                          s->head = s->tail = NULL;
  
                                          free(msg);
                                  } else if (msg->type == SEEK) {
                                          flush_processed(s->head);
                                          s->queue_len = 0;
                                          s->head = s->tail = NULL;
  
                                          struct seek_data *data = (struct seek_data *) msg->data;
                                          free(msg);
                                          if(data->whence == IMPORT_SEEK_CUR) {
                                                  index += data->offset;
                                          } else if (data->whence == IMPORT_SEEK_SET) {
                                                  index = data->offset;
                                          } else if (data->whence == IMPORT_SEEK_END) {
                                                  index = s->count + data->offset;
                                          }
                                          index = min(max(0, index), s->count - 1);
                                          printf("Current index: frame %d\n", index);
                                          free(data);
                                  } else {
                                          fprintf(stderr, "Unknown message type: %d!\n", msg->type);
                                          abort();
                                  }
                          }
                  }
  
                  /// @todo are these checks necessary?
                  index = min(max(0, index), s->count - 1);
  
                  struct video_reader_data data_reader[MAX_NUMBER_WORKERS];
                  task_result_handle_t task_handle[MAX_NUMBER_WORKERS];
  
                  int number_workers = s->video_reading_threads_count;
                  if (index + number_workers >= s->count) {
                          number_workers = s->count - index;
                  }
                  // run workers
                  for (int i = 0; i < number_workers; ++i) {
                          struct video_reader_data *data =
                                  &data_reader[i];
                          data->o_direct = s->o_direct;
                          data->tile_count = s->video_desc.tile_count;
                          data->tile_delim = s->tile_delim;
                          snprintf(data->file_name_prefix, sizeof(data->file_name_prefix),
                                          "%s/%08d", s->directory, index + i + 1);
                          strncpy(data->file_name_suffix,
                                          get_codec_file_extension(s->video_desc.color_spec),
                                          sizeof(data->file_name_suffix));
                          data->entry = NULL;
                          task_handle[i] = task_run_async(video_reader_callback, data);
                  }
  
                  // wait for workers to finish
                  for (int i = 0; i < number_workers; ++i) {
                          struct video_reader_data *data =
                                  (struct video_reader_data *)
                                  wait_task(task_handle[i]);
                          if (!data || data->entry == NULL)
                                  continue;
                          {
                                  unique_lock<mutex> lk(s->lock);
                                  if(s->head) {
                                          s->tail->next = data->entry;
                                          s->tail = data->entry;
                                  } else {
                                          s->head = s->tail = data->entry;
                                  }
                                  s->queue_len += 1;
  
                                  lk.unlock();
                                  s->boss_cv.notify_one();
                          }
                  }
                  index += number_workers;
          }
  
          return NULL;
  }
  
  static void reset_import(struct vidcap_import_state *s)
  {
          exit_reading_threads(s);
  
          s->finished = false;
  
          // clear audio state
          /// @todo
          /// This stuff is very ugly, rewrite it
          if (s->audio_state.has_audio) {
                  s->audio_state.played_samples = 0;
                  s->audio_state.samples_read = 0;
                  ring_buffer_flush(s->audio_state.data);
                  fseek(s->audio_state.file, 0L, SEEK_SET);
                  struct wav_metadata metadata;
                  read_wav_header(s->audio_state.file, &metadata); // skip metadata
          }
          s->frames_prev = s->frames = 0;
  
          if(pthread_create(&s->thread_id, NULL, reading_thread, (void *) s) != 0) {
                  fprintf(stderr, "Unable to create thread.\n");
                  /// @todo what to do here
                  abort();
          }
          if (s->audio_state.has_audio) {
                  if(pthread_create(&s->audio_state.thread_id, NULL, audio_reading_thread, (void *) s) != 0) {
                          fprintf(stderr, "Unable to create thread.\n");
                          abort();
                  }
          }
  }
  
  static void vidcap_import_dispose_video_frame(struct video_frame *frame) {
          free_entry((struct processed_entry *) frame->callbacks.dispose_udata);
          vf_free(frame);
  }
  
  static struct video_frame *
  vidcap_import_grab(void *state, struct audio_frame **audio)
  {
  	struct vidcap_import_state 	*s = (struct vidcap_import_state *) state;
          struct timeval cur_time;
          struct video_frame *ret;
          
          struct processed_entry *current = NULL;
  
          {
                  unique_lock<mutex> lk(s->lock);
                  if(s->queue_len == 0) {
                          if (s->finished == true && s->loop) {
                                  lk.unlock();
                                  log_msg(LOG_LEVEL_NOTICE, MOD_NAME "Rewinding the sequence.\n");
                                  reset_import(s);
                                  lk.lock();
                          }
                          if (s->finished == true && s->should_exit_at_end == true) {
                                  exit_uv(0);
                          }
  
                          s->boss_cv.wait_for(lk, duration<double>(2 * 1/s->video_desc.fps));
                  }
   
                  if(s->queue_len == 0) {
                          return NULL;
                  }
  
                  current = s->head;
                  assert(current != NULL);
  
                  s->head = s->head->next;
                  s->queue_len -= 1;
  
                  lk.unlock();
                  s->worker_cv.notify_one();
  
                  ret = vf_alloc_desc(s->video_desc);
                  ret->callbacks.dispose = vidcap_import_dispose_video_frame;
                  ret->callbacks.dispose_udata = current;
                  for (unsigned int i = 0; i < s->video_desc.tile_count; ++i) {
                          ret->tiles[i].data_len =
                                  current->tiles[i].data_len;
                          ret->tiles[i].data = current->tiles[i].data;
                  }
          }
  
  
          // audio
          if(s->audio_state.has_audio) {
                  unsigned long long int requested_samples = (unsigned long long int) (s->frames + 1) *
                          s->audio_frame.sample_rate / s->video_desc.fps - s->audio_state.played_samples;
                  if((int) (s->audio_state.played_samples + requested_samples) > s->audio_state.total_samples) {
                          requested_samples = s->audio_state.total_samples - s->audio_state.played_samples;
                  }
                  unsigned long long int requested_bytes = requested_samples * s->audio_frame.bps * s->audio_frame.ch_count;
                  if(requested_bytes) {
                          {
                                  requested_bytes = min<unsigned long long>(requested_bytes,s->audio_frame.max_size);
                                  unique_lock<mutex> lk(s->audio_state.lock);
                                  while(ring_get_current_size(s->audio_state.data) < (int) requested_bytes) {
                                          s->audio_state.boss_cv.wait(lk);;
                                  }
  
                                  int ret = ring_buffer_read(s->audio_state.data, s->audio_frame.data, requested_bytes);
  
                                  assert(ret == (int) requested_bytes);
  
                                  lk.unlock();
                                  s->audio_state.worker_cv.notify_one();
                          }
                  }
  
                  s->audio_frame.data_len = requested_bytes;
                  s->audio_state.played_samples += requested_samples;
                  *audio = &s->audio_frame;
          } else {
                  *audio = NULL;
          }
  
  
          gettimeofday(&cur_time, NULL);
          while(tv_diff_usec(cur_time, s->prev_time) < 1000000.0 / s->video_desc.fps) {
                  gettimeofday(&cur_time, NULL);
          }
          //tv_add_usec(&s->prev_time, 1000000.0 / s->frame->fps);
          s->prev_time = cur_time;
  
          double seconds = tv_diff(cur_time, s->t0);
          if (seconds >= 5) {
                  float fps  = (s->frames - s->frames_prev) / seconds;
                  log_msg(LOG_LEVEL_INFO, "[import] %d frames in %g seconds = %g FPS\n", s->frames - s->frames_prev, seconds, fps);
                  s->t0 = cur_time;
                  s->frames_prev = s->frames;
          }
  
          s->frames += 1;
  
  	return ret;
  }
  
  static const struct video_capture_info vidcap_import_info = {
          vidcap_import_probe,
          vidcap_import_init,
          vidcap_import_done,
          vidcap_import_grab,
          false
  };
  
  REGISTER_MODULE(import, &vidcap_import_info, LIBRARY_CLASS_VIDEO_CAPTURE, VIDEO_CAPTURE_ABI_VERSION);
  
  