1 files changed, 274 insertions, 366 deletions

diff --git a/src/main/extractor_ipc_gnu.c b/src/main/extractor_ipc_gnu.c
index b7f1824..7e7a086 100644
--- a/src/main/extractor_ipc_gnu.c
+++ b/src/main/extractor_ipc_gnu.c
@@ -1,6 +1,6 @@
 /*
      This file is part of libextractor.
-     (C) 2002, 2003, 2004, 2005, 2006, 2009, 2012 Vidyut Samanta and Christian Grothoff
+     (C) 2012 Vidyut Samanta and Christian Grothoff
 
      libextractor is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
@@ -17,24 +17,51 @@
      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
      Boston, MA 02111-1307, USA.
  */
-
+/**
+ * @file main/extractor_ipc_gnu.c
+ * @brief IPC with plugin for GNU/POSIX systems
+ * @author Christian Grothoff
+ */
 #include "platform.h"
 #include "plibc.h"
 #include "extractor.h"
+#include "extractor_datasource.h"
+#include "extractor_ipc.h"
 #include <dirent.h>
 #include <sys/types.h>
 #include <sys/wait.h>
 #include <sys/shm.h>
 #include <signal.h>
 
+/**
+ * Size of the channel buffer; determines largest IPC message that
+ * is going to be allowed.  FIXME: we might want to grow this
+ * buffer dynamically instead...
+ */
+#define CHANNEL_BUFFER_SIZE (1024 * 256)
 
 /**
- * Definition of an IPC communication channel with
- * some plugin.
+ * A shared memory resource (often shared with several
+ * other processes).
  */
-struct EXTRACTOR_Channel
+struct EXTRACTOR_SharedMemory
 {
   /**
+   * Pointer to the mapped region of the shm (covers the whole shm)
+   */ 
+  void *shm_ptr;
+
+  /**
+   * Page size. Mmap offset is a multiple of this number.
+   */
+  long allocation_granularity;
+
+  /**
+   * Allocated size of the shm
+   */ 
+  size_t shm_size;
+
+  /**
    * POSIX id of the shm into which data is uncompressed
    */ 
   int shm;
@@ -44,447 +71,328 @@ struct EXTRACTOR_Channel
    */ 
   char shm_name[MAX_SHM_NAME + 1];
 
+};
+
+
+/**
+ * Definition of an IPC communication channel with
+ * some plugin.
+ */
+struct EXTRACTOR_Channel
+{
+
   /**
-   * Pointer to the mapped region of the shm (covers the whole shm)
-   */ 
-  void *shm_ptr;
+   * Buffer for reading data from the plugin.
+   */
+  char data[CHANNEL_BUFFER_SIZE];
 
   /**
-   * Position within shm
-   */ 
-  int64_t shm_pos;
+   * Memory segment shared with this process.
+   */
+  struct EXTRACTOR_SharedMemory *shm;
 
   /**
-   * Allocated size of the shm
-   */ 
-  int64_t shm_size;
+   * Name of the plugin to use for this channel.
+   */
+  const char *short_libname;
 
   /**
-   * Number of bytes in shm (<= shm_size)
-   */ 
-  size_t shm_buf_size;
+   * Pipe used to communicate information to the plugin child process.
+   * NULL if not initialized.
+   */
+  int cpipe_in;
 
+  /**
+   * Number of valid bytes in the channel's buffer.
+   */
+  size_t size;
 
-};
+  /**
+   * Pipe used to read information about extracted meta data from
+   * the plugin child process.  -1 if not initialized.
+   */
+  int cpipe_out;
 
+  /**
+   * Process ID of the child process for this plugin. 0 for none.
+   */
+  pid_t cpid;
 
-/**
- * Opens a shared memory object (for later mmapping).
- * This is POSIX variant of the the plugin_open_* function. Shm is always memory-backed.
- * Closes a shm is already opened, closes it before opening a new one.
- *
- * @param plugin plugin context
- * @param shm_name name of the shm.
- * @return shm id (-1 on error). That is, the result of shm_open() syscall.
- */ 
-static int
-plugin_open_shm (struct EXTRACTOR_PluginList *plugin, 
-                 const char *shm_name)
-{
-  if (plugin->shm_id != -1)
-    close (plugin->shm_id);
-  plugin->shm_id = shm_open (shm_name, O_RDONLY, 0);
-  return plugin->shm_id;
-}
+};
 
 
 /**
- * Opens a file (for later mmapping).
- * This is POSIX variant of the plugin_open_* function.
- * Closes a file is already opened, closes it before opening a new one.
+ * Create a shared memory area.
  *
- * @param plugin plugin context
- * @param shm_name name of the file to open.
- * @return file id (-1 on error). That is, the result of open() syscall.
- */ 
-static int
-plugin_open_file (struct EXTRACTOR_PluginList *plugin, 
-                  const char *shm_name)
+ * @param size size of the shared area
+ * @return NULL on error
+ */
+struct EXTRACTOR_SharedMemory *
+EXTRACTOR_IPC_shared_memory_create_ (size_t size)
 {
-  if (plugin->shm_id != -1)
-    close (plugin->shm_id);
-  plugin->shm_id = open (shm_name, O_RDONLY, 0);
-  return plugin->shm_id;
+  struct EXTRACTOR_SharedMemory *shm;
+  const char *tpath;
+
+  if (NULL == (shm = malloc (sizeof (struct EXTRACTOR_SharedMemory))))
+    return NULL;
+#if SOMEBSD
+  /* this works on FreeBSD, not sure about others... */
+  tpath = getenv ("TMPDIR");
+  if (tpath == NULL)
+    tpath = "/tmp/";
+#else
+  tpath = "/"; /* Linux */
+#endif 
+  snprintf (shm->shm_name,
+            MAX_SHM_NAME, 
+            "%slibextractor-shm-%u-%u", 
+            tpath, getpid (),
+            (unsigned int) RANDOM());
+  if (-1 == (shm->shm_id = shm_open (shm->shm_name,
+                                     O_RDWR | O_CREAT, S_IRUSR | S_IWUSR)))
+    {
+      free (shm);
+      return NULL;
+    }
+  if ( (0 != ftruncate (shm->shm_id, size)) ||
+       (NULL == (shm->shm_ptr = mmap (NULL, size, 
+                                      PROT_WRITE, MAP_SHARED, 
+                                      shm->shm_id, 0))) ||
+       (((void*) -1) == shm->shm_ptr) )
+  {
+    (void) close (shm->shm_id);
+    (void) shm_unlink (shm->shm_name);
+    free (shm);
+    return NULL;
+  }
+  shm->shm_size = size;
+  return shm; 
 }
 
 
 /**
- * Initializes an extracting session for a plugin.
- *   opens the file/shm (only in OPMODE_FILE)
- *   sets shm_ptr to NULL (unmaps it, if it was mapped)
- *   sets position to 0
- *   initializes file size to 'fsize' (may be -1)
- *   sets seek request to 0
+ * Destroy shared memory area.
  *
- * @param plugin plugin context
- * @param operation_mode the mode of operation (OPMODE_*)
- * @param fsize size of the source file (may be -1)
- * @param shm_name name of the shm or file to open
- * @return 0 on success, non-0 on error.
- */ 
-static int
-init_state_method (struct EXTRACTOR_PluginList *plugin, 
-                   uint8_t operation_mode, 
-                   int64_t fsize, 
-                   const char *shm_name)
+ * @param shm memory area to destroy
+ * @return NULL on error
+ */
+void
+EXTRACTOR_IPC_shared_memory_destroy_ (struct EXTRACTOR_SharedMemory *shm)
 {
-  plugin->seek_request = 0;
-  if (plugin->shm_ptr != NULL)
-    munmap (plugin->shm_ptr, plugin->map_size);
-  plugin->shm_ptr = NULL;
-  if (operation_mode == OPMODE_FILE)
-  {
-    if (-1 == plugin_open_file (plugin, shm_name))
-      return 1;
-  }
-  else if (-1 == plugin_open_shm (plugin, shm_name))
-    return 1;
-  plugin->fsize = fsize;
-  plugin->shm_pos = 0;
-  plugin->fpos = 0;
-  return 0;
+  munmap (shm->shm_ptr, shm->map_size);
+  (void) close (plugin->shm_id);
+  (void) shm_unlink (shm->shm_name);
+  free (shm);
 }
 
 
 /**
- * Deinitializes an extracting session for a plugin.
- *   unmaps shm_ptr (if was mapped)
- *   closes file/shm (if it was opened)
- *   sets map size and shm_ptr to NULL.
+ * Initialize shared memory area from data source.
  *
- * @param plugin plugin context
- */ 
-static void
-discard_state_method (struct EXTRACTOR_PluginList *plugin)
+ * @param shm memory area to initialize
+ * @param ds data source to use for initialization
+ * @param off offset to use in data source
+ * @param size number of bytes to copy
+ * @return -1 on error, otherwise number of bytes copied
+ */
+ssize_t
+EXTRACTOR_IPC_shared_memory_set_ (struct EXTRACTOR_SharedMemory *shm,
+                                  struct EXTRACTOR_Datasource *ds,
+                                  uint64_t off,
+                                  size_t size)
 {
-  if (plugin->shm_ptr != NULL && plugin->map_size > 0)
-    munmap (plugin->shm_ptr, plugin->map_size);
-  if (plugin->shm_id != -1)
-    close (plugin->shm_id);
-  plugin->shm_id = -1;
-  plugin->map_size = 0;
-  plugin->shm_ptr = NULL;
+  if (-1 ==
+      EXTRACTOR_datasource_seek_ (ds, off, SEEK_SET))
+    return -1;
+  if (size > shm->map_size)
+    size = shm->map_size;
+  return EXTRACTOR_datasource_read_ (ds,
+                                     shm->shm_ptr,
+                                     size);
 }
 
 
-
 /**
- * Start the process for the given plugin.
+ * Create a channel to communicate with a process wrapping
+ * the plugin of the given name.  Starts the process as well.
+ *
+ * @param short_libname name of the plugin
+ * @param shm memory to share with the process
+ * @return NULL on error, otherwise IPC channel
  */ 
-static void
-start_process (struct EXTRACTOR_PluginList *plugin)
+struct EXTRACTOR_Channel *
+EXTRACTOR_IPC_channel_create_ (const char *short_libname,
+                               struct EXTRACTOR_SharedMemory *shm)
 {
+  struct EXTRACTOR_Channel *channel;
   int p1[2];
   int p2[2];
   pid_t pid;
   int status;
 
-  switch (plugin->flags)
-  {
-  case EXTRACTOR_OPTION_DEFAULT_POLICY:
-    if (-1 != plugin->cpid && 0 != plugin->cpid)
-      return;
-    break;
-  case EXTRACTOR_OPTION_OUT_OF_PROCESS_NO_RESTART:
-    if (0 != plugin->cpid)
-      return;
-    break;
-  case EXTRACTOR_OPTION_IN_PROCESS:
-    return;
-    break;
-  case EXTRACTOR_OPTION_DISABLED:
-    return;
-    break;
-  }
-
-  plugin->cpid = -1;
+  if (NULL == (channel = malloc (sizeof (struct EXTRACTOR_Channel))))
+    return NULL;
+  channel->shm = shm;
+  channel->short_libname = short_libname;
   if (0 != pipe (p1))
-  {
-    plugin->flags = EXTRACTOR_OPTION_DISABLED;
-    return;
-  }
+    {
+      free (channel);
+      return NULL;
+    }
   if (0 != pipe (p2))
-  {
-    close (p1[0]);
-    close (p1[1]);
-    plugin->flags = EXTRACTOR_OPTION_DISABLED;
-    return;
-  }
+    {
+      (void) close (p1[0]);
+      (void) close (p1[1]);
+      free (channel);
+      return;
+    }
   pid = fork ();
-  plugin->cpid = pid;
   if (pid == -1)
-  {
-    close (p1[0]);
-    close (p1[1]);
-    close (p2[0]);
-    close (p2[1]);
-    plugin->flags = EXTRACTOR_OPTION_DISABLED;
-    return;
-  }
-  if (pid == 0)
-  {
-    close (p1[1]);
-    close (p2[0]);
-    plugin_main (plugin, p1[0], p2[1]);
-    _exit (0);
-  }
-  close (p1[0]);
-  close (p2[1]);
-  plugin->cpipe_in = fdopen (p1[1], "w");
-  if (plugin->cpipe_in == NULL)
-  {
-    perror ("fdopen");
-    (void) kill (plugin->cpid, SIGKILL);
-    waitpid (plugin->cpid, &status, 0);
-    close (p1[1]);
-    close (p2[0]);
-    plugin->cpid = -1;
-    plugin->flags = EXTRACTOR_OPTION_DISABLED;
-    return;
-  }
-  plugin->cpipe_out = p2[0];
-}
-
-
-/**
- * Stop the child process of this plugin.
- */
-static void
-stop_process (struct EXTRACTOR_PluginList *plugin)
-{
-  int status;
-
-#if DEBUG
-  if (plugin->cpid == -1)
-    fprintf (stderr,
-             "Plugin `%s' choked on this input\n",
-             plugin->short_libname);
-#endif
-  if ( (plugin->cpid == -1) ||
-       (plugin->cpid == 0) )
-    return;
-  kill (plugin->cpid, SIGKILL);
-  waitpid (plugin->cpid, &status, 0);
-  plugin->cpid = -1;
-  close (plugin->cpipe_out);
-  fclose (plugin->cpipe_in);
-  plugin->cpipe_out = -1;
-  plugin->cpipe_in = NULL;
-
-  if (plugin->flags != EXTRACTOR_OPTION_DEFAULT_POLICY)
-    plugin->flags = EXTRACTOR_OPTION_DISABLED;
-
-  plugin->seek_request = -1;
-}
-
-
-static int
-write_plugin_data (const struct EXTRACTOR_PluginList *plugin)
-{
-  /* This function is only necessary on W32. On POSIX
-   * systems plugin inherits its own data from the parent */
-  return 0;
-}
-
-
-/**
- * Initializes an extracting session for a plugin.
- *   opens the file/shm (only in OPMODE_FILE)
- *   sets shm_ptr to NULL (unmaps it, if it was mapped)
- *   sets position to 0
- *   initializes file size to 'fsize' (may be -1)
- *   sets seek request to 0
- *
- * @param plugin plugin context
- * @param operation_mode the mode of operation (OPMODE_*)
- * @param fsize size of the source file (may be -1)
- * @param shm_name name of the shm or file to open
- * @return 0 on success, non-0 on error.
- */ 
-static int
-init_state_method (struct EXTRACTOR_PluginList *plugin, 
-                   uint8_t operation_mode, 
-                   int64_t fsize, 
-                   const char *shm_name)
-{
-  plugin->seek_request = 0;
-  if (plugin->shm_ptr != NULL)
-    munmap (plugin->shm_ptr, plugin->map_size);
-  plugin->shm_ptr = NULL;
-  if (operation_mode == OPMODE_FILE)
-  {
-    if (-1 == plugin_open_file (plugin, shm_name))
-      return 1;
-  }
-  else if (-1 == plugin_open_shm (plugin, shm_name))
-    return 1;
-  plugin->fsize = fsize;
-  plugin->shm_pos = 0;
-  plugin->fpos = 0;
-  return 0;
+    {
+      (void) close (p1[0]);
+      (void) close (p1[1]);
+      (void) close (p2[0]);
+      (void) close (p2[1]);
+      free (channel);
+      return NULL;
+    }
+  if (0 == pid)
+    {
+      (void) close (p1[1]);
+      (void) close (p2[0]);
+      EXTRACTOR_plugin_main_ (short_libname, p1[0], p2[1]);
+      _exit (0);
+    }
+  (void) close (p1[0]);
+  (void) close (p2[1]);
+  channel->cpipe_in = p1[1];
+  channel->cpipe_out = p2[0];
+  channel->cpid = pid;
+  return channel;
 }
 
 
 /**
- * Setup a shared memory segment.
+ * Destroy communication channel with a plugin/process.  Also
+ * destroys the process.
  *
- * @param ptr set to the location of the shm segment
- * @param shmid where to store the shm ID
- * @param fn name of the shared segment
- * @param fn_size size available in fn
- * @param size number of bytes to allocated for the segment
- * @return 0 on success
+ * @param channel channel to communicate with the plugin
  */
-static int
-make_shm_posix (void **ptr, 
-                int *shmid, 
-                char *fn, 
-                size_t fn_size, size_t size)
+void
+EXTRACTOR_IPC_channel_destroy_ (struct EXTRACTOR_Channel *channel)
 {
-  const char *tpath;
-#if SOMEBSD
-  /* this works on FreeBSD, not sure about others... */
-  tpath = getenv ("TMPDIR");
-  if (tpath == NULL)
-    tpath = "/tmp/";
-#else
-  tpath = "/"; /* Linux */
-#endif 
-  snprintf (fn, fn_size, "%slibextractor-shm-%u-%u", tpath, getpid(),
-      (unsigned int) RANDOM());
-  *shmid = shm_open (fn, O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
-  *ptr = NULL;
-  if (-1 == *shmid)
-    return 1;
-  if ((0 != ftruncate (*shmid, size)) ||
-      (NULL == (*ptr = mmap (NULL, size, PROT_WRITE, MAP_SHARED, *shmid, 0))) ||
-      (*ptr == (void*) -1) )
-  {
-    close (*shmid);
-    *shmid = -1;
-    shm_unlink (fn);
-    return 1;
-  }
-  return 0;
-}
-
+  int status;
 
-static void
-destroy_shm_posix (void *ptr, int shm_id, size_t size, char *shm_name)
-{
-  if (NULL != ptr)
-    munmap (ptr, size);
-  if (shm_id != -1)
-    close (shm_id);
-  shm_unlink (shm_name);
+  (void) kill (channel->cpid, SIGKILL);
+  (void) waitpid (channel->cpid, &status, 0);
+  (void) close (channel->cpipe_out);
+  (void) close (channel->cpipe_in);
+  free (channel);
 }
 
 
 /**
- * Receive 'size' bytes from plugin, store them in 'buf'
+ * Send data via the given IPC channel (blocking).
  *
- * @param plugin plugin context
- * @param buf buffer to fill
- * @param size number of bytes to read
- * @return number of bytes read, 0 on EOS, < 0 on error
+ * @param channel channel to communicate with the plugin
+ * @param buf data to send
+ * @param size number of bytes in buf to send
+ * @return -1 on error, number of bytes sent on success
+ *           (never does partial writes)
  */
-static int
-plugin_read (struct EXTRACTOR_PluginList *plugin, 
-             void *buf, 
-             size_t size)
+ssize_t
+EXTRACTOR_IPC_channel_send_ (struct EXTRACTOR_Channel *channel,
+                             const void *data,
+                             size_t size)
 {
-  char *rb = buf;
-  ssize_t read_result;
-  size_t read_count = 0;
+  const char *cdata = data;
+  size_t off = 0;
+  ssize_t ret;
 
-  while (read_count < size)
-  {
-    read_result = read (plugin->cpipe_out, 
-                        &rb[read_count], size - read_count);
-    if (read_result <= 0)
-      return read_result;
-    read_count += read_result;
-  }
-  return read_count;
+  while (off < size)
+    {
+      ret = write (channel->cpipe_in, &cdata[off], size - off);
+      if (ret <= 0)
+        return -1;
+      off += ret;
+    }
+  return size;
 }
 
 
 /**
+ * Receive data from any of the given IPC channels (blocking).
  * Wait for one of the plugins to reply.
- * Selects on plugin output pipes, runs receive_reply()
+ * Selects on plugin output pipes, runs 'receive_reply'
  * on each activated pipe until it gets a seek request
  * or a done message. Called repeatedly by the user until all pipes are dry or
  * broken.
  *
- * @param plugins to select upon
- * @param proc metadata callback
- * @param proc_cls callback cls
- * @return number of dry/broken pipes since last call, -1 on error or if no
- *         plugins reply in 10 seconds.
+ * @param channels array of channels, channels that break may be set to NULL
+ * @param num_channels length of the 'channels' array
+ * @param proc function to call to process messages (may be called
+ *             more than once)
+ * @param proc_cls closure for 'proc'
+ * @return -1 on error, 1 on success
  */
-static int
-wait_for_reply (struct EXTRACTOR_PluginList *plugins,
-                EXTRACTOR_MetaDataProcessor proc, void *proc_cls)
+int
+EXTRACTOR_IPC_channel_recv_ (struct EXTRACTOR_Channel **channels,
+                             unsigned int num_channels,
+                             EXTRACTOR_ChannelMessageProcessor proc,
+                             void *proc_cls)
 {
-  int ready;
-  int result;
   struct timeval tv;
   fd_set to_check;
-  int highest = 0;
-  int read_result;
-  struct EXTRACTOR_PluginList *ppos;
+  int max;
+  unsigned int i;
+  struct EXTRACTOR_Channel *channel;
+  ssize_t ret;
+  ssize_t iret;
 
   FD_ZERO (&to_check);
-  for (ppos = plugins; NULL != ppos; ppos = ppos->next)
+  max = -1;
+  for (i=0;i<num_channels;i++)
     {
-      switch (ppos->flags)
-        {
-        case EXTRACTOR_OPTION_DEFAULT_POLICY:
-        case EXTRACTOR_OPTION_OUT_OF_PROCESS_NO_RESTART:
-          if (ppos->seek_request == -1)
-            continue;
-          FD_SET (ppos->cpipe_out, &to_check);
-          if (highest < ppos->cpipe_out)
-            highest = ppos->cpipe_out;
-          break;
-        case EXTRACTOR_OPTION_IN_PROCESS:
-          break;
-        case EXTRACTOR_OPTION_DISABLED:
-          break;
-        }
+      channel = channels[i];
+      if (NULL == channel)
+        continue;
+      FD_SET (channel->cpipe_out, &to_check);
+      if (max < channel->cpipe_out)
+        max = channel->cpipe_out;
     }
-  
   tv.tv_sec = 10;
   tv.tv_usec = 0;
-  ready = select (highest + 1, &to_check, NULL, NULL, &tv);
-  if (ready <= 0)
-    /* an error or timeout -> something's wrong or all plugins hung up */
-    return -1;
-
-  result = 0;
-  for (ppos = plugins; NULL != ppos; ppos = ppos->next)
+  if (-1 == select (max + 1, &to_check, NULL, NULL, &tv))
     {
-      switch (ppos->flags)
+      /* an error or timeout -> something's wrong or all plugins hung up */
+      return -1;
+    }
+  for (i=0;i<num_channels;i++)
+    {
+      channel = channels[i];
+      if (NULL == channel)
+        continue;
+      if (! FD_ISSET (channel->cpipe_out, &to_check))
+        continue;
+      if ( (-1 == (iret = read (channel->cpipe_out,
+                                &channel->data[channel->size],
+                                CHANNEL_BUFFER_SIZE - channel->size)) ) ||
+           (ret = EXTRACTOR_IPC_process_reply_ (channel->data, 
+                                                channel->size + iret, 
+                                                proc, proc_cls)) )
+        {
+          EXTRACTOR_IPC_channel_destroy (channel);
+          channels[i] = NULL;
+        }
+      else
         {
-        case EXTRACTOR_OPTION_DEFAULT_POLICY:
-        case EXTRACTOR_OPTION_OUT_OF_PROCESS_NO_RESTART:
-          if (ppos->seek_request == -1)
-            continue;
-          if (FD_ISSET (ppos->cpipe_out, &to_check))
-            {
-              read_result = receive_reply (ppos, proc, proc_cls);
-              if (read_result < 0)
-                {
-                  stop_process (ppos);
-                }
-              result += 1;
-            }
-          break;
-        case EXTRACTOR_OPTION_IN_PROCESS:
-          break;
-        case EXTRACTOR_OPTION_DISABLED:
-          break;
+          memmove (channel->data,
+                   &channel->data[ret],
+                   channel->size + iret - ret);
+          channel->size = channel->size + iret - ret;
         }
     }
-  return result;
+  return 1;
 }
+
+/* end of extractor_ipc_gnu.c */