[GNUnet-SVN] r22880 - Extractor/src/main

[gnunet]

Author: grothoff
Date: 2012-07-24 23:56:19 +0200 (Tue, 24 Jul 2012)
New Revision: 22880

Modified:
   Extractor/src/main/extractor.c
Log:
stuff

Modified: Extractor/src/main/extractor.c
===================================================================
--- Extractor/src/main/extractor.c      2012-07-24 21:45:25 UTC (rev 22879)
+++ Extractor/src/main/extractor.c      2012-07-24 21:56:19 UTC (rev 22880)
@@ -39,398 +39,6 @@
 
 #if 0
 /**
- * Open a file
- */
-static int 
-file_open (const char *filename, int oflag, ...)
-{
-  int mode;
-  const char *fn;
-#ifdef MINGW
-  char szFile[_MAX_PATH + 1];
-  long lRet;
-
-  if ((lRet = plibc_conv_to_win_path(filename, szFile)) != ERROR_SUCCESS)
-  {
-    errno = ENOENT;
-    SetLastError(lRet);
-    return -1;
-  }
-  fn = szFile;
-#else
-  fn = filename;
-#endif
-  mode = 0;
-#ifdef MINGW
-  /* Set binary mode */
-  mode |= O_BINARY;
-#endif
-  return OPEN(fn, oflag, mode);
-}
-
-
-/**
- * Initializes plugin state. Calls init_state_method()
- * directly or indirectly.
- *
- * @param plugin plugin to initialize
- * @param operation_mode operation mode
- * @param shm_name name of the shm/file
- * @param fsize file size (may be -1)
- */
-static void
-init_plugin_state (struct EXTRACTOR_PluginList *plugin, 
-                  uint8_t operation_mode, 
-                  const char *shm_name, int64_t fsize)
-{
-  int write_result;
-  int init_state_size;
-  unsigned char *init_state;
-  int t;
-  size_t shm_name_len = strlen (shm_name) + 1;
-  
-  init_state_size = 1 + sizeof (size_t) + shm_name_len + sizeof (uint8_t) + 
sizeof (int64_t);
-  plugin->operation_mode = operation_mode;
-  switch (plugin->flags)
-  {
-  case EXTRACTOR_OPTION_DEFAULT_POLICY:
-  case EXTRACTOR_OPTION_OUT_OF_PROCESS_NO_RESTART:
-    init_state = malloc (init_state_size);
-    if (init_state == NULL)
-    {
-      stop_process (plugin);
-      return;
-    }
-    t = 0;
-    init_state[t] = MESSAGE_INIT_STATE;
-    t += 1;
-    memcpy (&init_state[t], &operation_mode, sizeof (uint8_t));
-    t += sizeof (uint8_t);
-    memcpy (&init_state[t], &fsize, sizeof (int64_t));
-    t += sizeof (int64_t);
-    memcpy (&init_state[t], &shm_name_len, sizeof (size_t));
-    t += sizeof (size_t);
-    memcpy (&init_state[t], shm_name, shm_name_len);
-    t += shm_name_len;
-    write_result = plugin_write (plugin, init_state, init_state_size);
-    free (init_state);
-    if (write_result < init_state_size)
-    {
-      stop_process (plugin);
-      return;
-    }
-    plugin->seek_request = 0;
-    break;
-  case EXTRACTOR_OPTION_IN_PROCESS:
-    init_state_method (plugin, operation_mode, fsize, shm_name);
-    return;
-    break;
-  case EXTRACTOR_OPTION_DISABLED:
-    return;
-    break;
-  }
-}
-
-
-/**
- * Discards plugin state. Calls discard_state_method()
- * directly or indirectly.
- *
- * @param plugin plugin to initialize
- */
-static void
-discard_plugin_state (struct EXTRACTOR_PluginList *plugin)
-{
-  int write_result;
-  unsigned char discard_state = MESSAGE_DISCARD_STATE;
-
-  switch (plugin->flags)
-  {
-  case EXTRACTOR_OPTION_DEFAULT_POLICY:
-  case EXTRACTOR_OPTION_OUT_OF_PROCESS_NO_RESTART:
-    /* This is somewhat clumsy, but it's the only stop-indicating
-     * non-W32/POSIX-specific field i could think of...
-     */
-    if (plugin->cpipe_out != -1)
-    {
-      write_result = plugin_write (plugin, &discard_state, 1);
-      if (write_result < 1)
-      {
-        stop_process (plugin);
-        return;
-      }
-    }
-    break;
-  case EXTRACTOR_OPTION_IN_PROCESS:
-    discard_state_method (plugin);
-    return;
-    break;
-  case EXTRACTOR_OPTION_DISABLED:
-    return;
-    break;
-  }
-}
-
-
-/**
- * Forces plugin to move the buffer window to 'pos'.
- *
- * @param plugin plugin context
- * @param pos position to move to
- * @param want_start 1 if the caller is interested in the beginning of the
- *        window, 0 if the caller is interested in its end. Window position
- *        must be aligned to page size, and this parameter controls the
- *        direction of window shift. 0 is used mostly by SEEK_END.
- * @return 0 on success, -1 on error
- */
-static int
-pl_pick_next_buffer_at (struct EXTRACTOR_PluginList *plugin,
-                       int64_t pos,
-                       uint8_t want_start)
-{
-  if (plugin->operation_mode == OPMODE_MEMORY)
-  {
-    int64_t old_pos;
-    int64_t gran_fix;
-#if !WINDOWS
-    if (plugin->shm_ptr != NULL)
-      munmap (plugin->shm_ptr, plugin->map_size);
-#else
-    if (plugin->shm_ptr != NULL)
-      UnmapViewOfFile (plugin->shm_ptr);
-#endif
-    plugin->shm_ptr = NULL;
-    old_pos = plugin->fpos + plugin->shm_pos;
-    if (pos < 0)
-      pos = 0;
-    if (pos > plugin->fsize)
-      pos = plugin->fsize - 1;
-    plugin->fpos = pos;
-    plugin->map_size = MAX_READ;
-    plugin->shm_pos = old_pos - plugin->fpos;
-    if (want_start)
-      gran_fix = -1 * (plugin->fpos % plugin->allocation_granularity);
-    else
-    {
-      gran_fix = plugin->fpos % plugin->allocation_granularity;
-      if (gran_fix > 0)
-        gran_fix = plugin->allocation_granularity - gran_fix;
-    }
-    if (plugin->fpos + gran_fix + plugin->map_size > plugin->fsize)
-      plugin->map_size = plugin->fsize - plugin->fpos - gran_fix;
-    plugin->fpos += gran_fix;
-#if !WINDOWS
-    if ((-1 == plugin->shm_id) ||
-        (NULL == (plugin->shm_ptr = mmap (NULL, plugin->map_size, PROT_READ, 
MAP_SHARED, plugin->shm_id, plugin->fpos))) ||
-        (plugin->shm_ptr == (void *) -1))
-    {
-      return -1;
-    }
-#else
-    LARGE_INTEGER off;
-    off.QuadPart = plugin->fpos;
-    if ((plugin->map_handle == 0) ||
-       (NULL == (plugin->shm_ptr = MapViewOfFile (plugin->map_handle, 
FILE_MAP_READ, off.HighPart, off.LowPart, plugin->map_size))))
-    {
-      DWORD err = GetLastError ();
-      return -1;
-    }
-#endif
-    plugin->shm_pos -= gran_fix;
-    return 0;
-  }
-  if (plugin->operation_mode == OPMODE_FILE)
-  {
-    int64_t old_pos;
-    int64_t gran_fix;
-#if !WINDOWS
-    if (plugin->shm_ptr != NULL)
-      munmap (plugin->shm_ptr, plugin->map_size);
-#else
-    if (plugin->shm_ptr != NULL)
-      UnmapViewOfFile (plugin->shm_ptr);
-#endif
-    plugin->shm_ptr = NULL;
-    old_pos = plugin->fpos + plugin->shm_pos;
-    if (pos < 0)
-      pos = 0;
-    if (pos > plugin->fsize)
-      pos = plugin->fsize - 1;
-    plugin->fpos = pos;
-    plugin->map_size = MAX_READ;
-    plugin->shm_pos = old_pos - plugin->fpos;
-    if (want_start)
-      gran_fix = -1 * (plugin->fpos % plugin->allocation_granularity);
-    else
-    {
-      gran_fix = plugin->fpos % plugin->allocation_granularity;
-      if (gran_fix > 0)
-        gran_fix = plugin->allocation_granularity - gran_fix;
-    }
-    if (plugin->fpos + gran_fix + plugin->map_size > plugin->fsize)
-      plugin->map_size = plugin->fsize - plugin->fpos - gran_fix;
-    plugin->fpos += gran_fix;
-#if !WINDOWS
-    if ((-1 == plugin->shm_id) ||
-        (NULL == (plugin->shm_ptr = mmap (NULL, plugin->map_size, PROT_READ, 
MAP_SHARED, plugin->shm_id, plugin->fpos))) ||
-        (plugin->shm_ptr == (void *) -1))
-    {
-      return -1;
-    }
-#else
-    LARGE_INTEGER off;
-    off.QuadPart = plugin->fpos;
-    if ((plugin->map_handle == 0) ||
-       (NULL == (plugin->shm_ptr = MapViewOfFile (plugin->map_handle, 
FILE_MAP_READ, off.HighPart, off.LowPart, plugin->map_size))))
-    {
-      DWORD err = GetLastError ();
-      return -1;
-    }
-#endif
-    plugin->shm_pos -= gran_fix;
-    return 0;
-  }
-  if (plugin->operation_mode == OPMODE_DECOMPRESS)
-  {
-    if (plugin->pipe_in != 0)
-    {
-      int64_t old_pos;
-      old_pos = plugin->fpos + plugin->shm_pos;
-      plugin->seek_request = pos;
-      /* Recourse into request loop to wait for shm update */
-      while (plugin->fpos != pos)
-      {
-        plugin->waiting_for_update = 1;
-        if (process_requests (plugin) < 0)
-          return -1;
-        plugin->waiting_for_update = 0;
-      }
-      plugin->shm_pos = old_pos - plugin->fpos;
-    }
-    else
-    {
-      if (pos < plugin->fpos)
-      {
-        if (1 != cfs_reset_stream (plugin->pass_cfs))
-          return -1;
-      }
-      while (plugin->fpos < pos && plugin->fpos >= 0)
-        plugin->fpos = cfs_seek (plugin->pass_cfs, pos);
-      plugin->fsize = ((struct CompressedFileSource 
*)plugin->pass_cfs)->uncompressed_size;
-      plugin->shm_pos = pos - plugin->fpos;
-    }
-    return 0;
-  }
-  return -1;
-}
-
-
-
-
-/**
- * Transmits information about updated shm to plugin.
- * For OPMODE_DECOMPRESS only.
- *
- * @param plugin plugin context
- * @param position current absolute position in uncompressed stream
- * @param map_size number of bytes that are available in shm
- * @param fsize total size of the uncompressed stream (might be -1)
- * @param operation_mode mode of operation
- * @return 0 on success, 1 on error
- */
-static int
-give_shm_to_plugin (struct EXTRACTOR_PluginList *plugin, 
-                   int64_t position,
-                   size_t map_size, int64_t fsize, 
-                   uint8_t operation_mode)
-{
-  int write_result;
-  int updated_shm_size = 1 + sizeof (int64_t) + sizeof (size_t) + sizeof 
(int64_t);
-  unsigned char updated_shm[updated_shm_size];
-  int t = 0;
- 
-  updated_shm[t] = MESSAGE_UPDATED_SHM;
-  t++;
-  memcpy (&updated_shm[t], &position, sizeof (int64_t));
-  t += sizeof (int64_t);
-  memcpy (&updated_shm[t], &map_size, sizeof (size_t));
-  t += sizeof (size_t);
-  memcpy (&updated_shm[t], &fsize, sizeof (int64_t));
-  t += sizeof (int64_t);
-  switch (plugin->flags)
-    {
-    case EXTRACTOR_OPTION_DEFAULT_POLICY:
-    case EXTRACTOR_OPTION_OUT_OF_PROCESS_NO_RESTART:
-      if (operation_mode == OPMODE_DECOMPRESS)
-       {
-         if (plugin->seek_request < 0)
-           return 0;
-         write_result = plugin_write (plugin, updated_shm, updated_shm_size);
-         if (write_result < updated_shm_size)
-           {
-             stop_process (plugin);
-             return 0;
-           }
-       }
-      return 1;
-    case EXTRACTOR_OPTION_IN_PROCESS:
-      if (operation_mode == OPMODE_DECOMPRESS)
-       {
-         plugin->fpos = position;
-         plugin->map_size = map_size;
-         plugin->fsize = fsize;
-       }
-      return 0;
-    case EXTRACTOR_OPTION_DISABLED:
-      return 0;
-    default:
-      return 1;
-    }
-}
-
-
-/**
- * Calls _extract_method of in-process plugin.
- *
- * @param plugin plugin context
- * @param shm_ptr pointer to the data buffer
- * @param proc metadata callback
- * @param proc_cls callback cls
- */
-static void
-ask_in_process_plugin (struct EXTRACTOR_PluginList *plugin,
-                      void *shm_ptr,
-                      EXTRACTOR_MetaDataProcessor proc, void *proc_cls)
-{
-  int extract_reply;
-
-  switch (plugin->flags)
-    {
-    case EXTRACTOR_OPTION_DEFAULT_POLICY:
-    case EXTRACTOR_OPTION_OUT_OF_PROCESS_NO_RESTART:
-      return;
-    case EXTRACTOR_OPTION_IN_PROCESS:
-      if (plugin->seek_request >= 0)
-       {
-         plugin->shm_ptr = shm_ptr;
-         extract_reply = plugin->extract_method (plugin, proc, proc_cls);
-         /* Don't leak errno from the extract method */
-         errno = 0;
-         if (1 == extract_reply)
-           plugin->seek_request = -1;
-       }
-      break;
-    case EXTRACTOR_OPTION_DISABLED:
-      return;
-      break;
-    }
-}
-
-
-
-
-
-/**
  * Checks the seek requests that plugins made, finds the one with
  * smallest offset from the beginning of the stream, and satisfies it.
  * 
@@ -476,22 +84,6 @@
     }
   return cfs_seek (cfs, min_pos);
 }
-
-
-static void
-load_in_process_plugin (struct EXTRACTOR_PluginList *plugin)
-{
-  switch (plugin->flags)
-    {
-    case EXTRACTOR_OPTION_DEFAULT_POLICY:
-    case EXTRACTOR_OPTION_OUT_OF_PROCESS_NO_RESTART:
-    case EXTRACTOR_OPTION_DISABLED:
-      break;
-    case EXTRACTOR_OPTION_IN_PROCESS:
-      EXTRACTOR_plugin_load_ (plugin);
-      break;
-    }
-}
 #endif
 
 
@@ -621,6 +213,15 @@
        done = 0;
       }
   }
+
+  /* run in-process plugins */
+  for (pos = plugins; NULL != pos; pos = pos->next)
+    {
+      if (EXTRACTOR_OPTION_IN_PROCESS != pos->flags)
+       continue;
+      // FIXME: initialize read/seek context...
+      // pos->extract_method (FIXME);
+    }
 }
 
 
@@ -668,7 +269,7 @@
     shm = EXTRACTOR_IPC_shared_memory_create_ (DEFAULT_SHM_SIZE);
   for (pos = plugins; NULL != pos; pos = pos->next)
     if ( (NULL == pos->shm) &&
-        (0 == (pos->flags & EXTRACTOR_OPTION_IN_PROCESS)) )
+        (EXTRACTOR_OPTION_IN_PROCESS == pos->flags) )
       {
        pos->shm = shm;
        pos->channel = EXTRACTOR_IPC_channel_create_ (pos,