gst/gstpluginloader.c

   1 /* GStreamer
   2  * Copyright (C) 2008 Jan Schmidt <jan.schmidt@sun.com>
   3  *
   4  * gstpluginloader.c: GstPluginLoader helper for loading plugin files
   5  * out of process.
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22
  23 #ifdef HAVE_CONFIG_H
  24 #  include "config.h"
  25 #endif
  26
  27 #ifndef G_OS_WIN32
  28 #include <sys/types.h>
  29 #include <sys/wait.h>
  30 #include <unistd.h>
  31 #endif
  32 #include <errno.h>
  33
  34 #include <gst/gst_private.h>
  35 #include <gst/gstconfig.h>
  36
  37 #include <gst/gstpoll.h>
  38 #include <gst/gstutils.h>
  39
  40 #include <gst/gstpluginloader.h>
  41 #include <gst/gstregistrychunks.h>
  42
  43 #define GST_CAT_DEFAULT GST_CAT_PLUGIN_LOADING
  44
  45 static GstPluginLoader *plugin_loader_new (GstRegistry * registry);
  46 static gboolean plugin_loader_free (GstPluginLoader * loader);
  47 static gboolean plugin_loader_load (GstPluginLoader * loader,
  48     const gchar * filename, off_t file_size, time_t file_mtime);
  49
  50 const GstPluginLoaderFuncs _priv_gst_plugin_loader_funcs = {
  51   plugin_loader_new, plugin_loader_free, plugin_loader_load
  52 };
  53
  54 typedef struct _PendingPluginEntry
  55 {
  56   guint32 tag;
  57   gchar *filename;
  58   off_t file_size;
  59   time_t file_mtime;
  60 } PendingPluginEntry;
  61
  62 struct _GstPluginLoader
  63 {
  64   GstRegistry *registry;
  65   GstPoll *fdset;
  66
  67   gboolean child_running;
  68   GPid child_pid;
  69   GstPollFD fd_w;
  70   GstPollFD fd_r;
  71
  72   gboolean is_child;
  73   gboolean got_plugin_details;
  74
  75   /* Transmit buffer */
  76   guint8 *tx_buf;
  77   guint tx_buf_size;
  78   guint tx_buf_write;
  79   guint tx_buf_read;
  80
  81   guint32 next_tag;
  82
  83   guint8 *rx_buf;
  84   guint rx_buf_size;
  85   gboolean rx_done;
  86   gboolean rx_got_sync;
  87
  88   /* Head and tail of the pending plugins list. List of
  89      PendingPluginEntry structs */
  90   GList *pending_plugins;
  91   GList *pending_plugins_tail;
  92 };
  93
  94 #define PACKET_EXIT 1
  95 #define PACKET_LOAD_PLUGIN 2
  96 #define PACKET_SYNC 3
  97 #define PACKET_PLUGIN_DETAILS 4
  98
  99 #define BUF_INIT_SIZE 512
 100 #define BUF_GROW_EXTRA 512
 101 #define HEADER_SIZE 8
 102 #define ALIGNMENT   (sizeof (void *))
 103
 104
 105 static gboolean gst_plugin_loader_spawn (GstPluginLoader * loader);
 106 static void put_packet (GstPluginLoader * loader, guint type, guint32 tag,
 107     const guint8 * payload, guint32 payload_len);
 108 static gboolean exchange_packets (GstPluginLoader * l);
 109 static gboolean plugin_loader_replay_pending (GstPluginLoader * l);
 110 static gboolean plugin_loader_load_and_sync (GstPluginLoader * l,
 111     PendingPluginEntry * entry);
 112 static void plugin_loader_create_blacklist_plugin (GstPluginLoader * l,
 113     PendingPluginEntry * entry);
 114 static void plugin_loader_cleanup_child (GstPluginLoader * loader);
 115
 116 static GstPluginLoader *
 117 plugin_loader_new (GstRegistry * registry)
 118 {
 119   GstPluginLoader *l = g_new0 (GstPluginLoader, 1);
 120
 121   if (registry)
 122     l->registry = gst_object_ref (registry);
 123   l->fdset = gst_poll_new (FALSE);
 124   gst_poll_fd_init (&l->fd_w);
 125   gst_poll_fd_init (&l->fd_r);
 126
 127   l->tx_buf_size = BUF_INIT_SIZE;
 128   l->tx_buf = g_malloc (BUF_INIT_SIZE);
 129
 130   l->next_tag = 0;
 131
 132   l->rx_buf_size = BUF_INIT_SIZE;
 133   l->rx_buf = g_malloc (BUF_INIT_SIZE);
 134
 135   return l;
 136 }
 137
 138 static gboolean
 139 plugin_loader_free (GstPluginLoader * loader)
 140 {
 141   GList *cur;
 142   gboolean got_plugin_details;
 143
 144   fsync (loader->fd_w.fd);
 145
 146   if (loader->child_running) {
 147     put_packet (loader, PACKET_EXIT, 0, NULL, 0);
 148
 149     /* Swap packets with the child until it exits cleanly */
 150     while (!loader->rx_done) {
 151       if (exchange_packets (loader) || loader->rx_done)
 152         continue;
 153
 154       if (!plugin_loader_replay_pending (loader))
 155         break;
 156       put_packet (loader, PACKET_EXIT, 0, NULL, 0);
 157     }
 158
 159     plugin_loader_cleanup_child (loader);
 160   } else {
 161     close (loader->fd_w.fd);
 162     close (loader->fd_r.fd);
 163   }
 164
 165   gst_poll_free (loader->fdset);
 166
 167   g_free (loader->rx_buf);
 168   g_free (loader->tx_buf);
 169
 170   if (loader->registry)
 171     gst_object_unref (loader->registry);
 172
 173   got_plugin_details = loader->got_plugin_details;
 174
 175   /* Free any pending plugin entries */
 176   cur = loader->pending_plugins;
 177   while (cur) {
 178     PendingPluginEntry *entry = (PendingPluginEntry *) (cur->data);
 179     g_free (entry->filename);
 180     g_free (entry);
 181
 182     cur = g_list_delete_link (cur, cur);
 183   }
 184
 185   g_free (loader);
 186
 187   return got_plugin_details;
 188 }
 189
 190 static gboolean
 191 plugin_loader_load (GstPluginLoader * loader, const gchar * filename,
 192     off_t file_size, time_t file_mtime)
 193 {
 194   gint len;
 195   PendingPluginEntry *entry;
 196
 197   if (!gst_plugin_loader_spawn (loader))
 198     return FALSE;
 199
 200   /* Send a packet to the child requesting that it load the given file */
 201   GST_LOG_OBJECT (loader->registry,
 202       "Sending file %s to child. tag %u", filename, loader->next_tag);
 203
 204   entry = g_new (PendingPluginEntry, 1);
 205   entry->tag = loader->next_tag++;
 206   entry->filename = g_strdup (filename);
 207   entry->file_size = file_size;
 208   entry->file_mtime = file_mtime;
 209   loader->pending_plugins_tail =
 210       g_list_append (loader->pending_plugins_tail, entry);
 211
 212   if (loader->pending_plugins == NULL)
 213     loader->pending_plugins = loader->pending_plugins_tail;
 214   else
 215     loader->pending_plugins_tail = g_list_next (loader->pending_plugins_tail);
 216
 217   len = strlen (filename);
 218   put_packet (loader, PACKET_LOAD_PLUGIN, entry->tag,
 219       (guint8 *) filename, len + 1);
 220
 221   if (!exchange_packets (loader)) {
 222     if (!plugin_loader_replay_pending (loader))
 223       return FALSE;
 224   }
 225
 226   return TRUE;
 227 }
 228
 229 static gboolean
 230 plugin_loader_replay_pending (GstPluginLoader * l)
 231 {
 232   GList *cur, *next;
 233
 234 restart:
 235   if (!gst_plugin_loader_spawn (l))
 236     return FALSE;
 237
 238   /* Load each plugin one by one synchronously until we find the
 239    * crashing one */
 240   while ((cur = l->pending_plugins)) {
 241     PendingPluginEntry *entry = (PendingPluginEntry *) (cur->data);
 242
 243     if (!plugin_loader_load_and_sync (l, entry)) {
 244       GST_INFO ("AHA! %s crashes on loading", entry->filename);
 245       /* Create dummy plugin entry to block re-scanning this file */
 246       plugin_loader_create_blacklist_plugin (l, entry);
 247       l->got_plugin_details = TRUE;
 248       /* Now remove this crashy plugin from the head of the list */
 249       l->pending_plugins = g_list_delete_link (cur, cur);
 250       if (l->pending_plugins == NULL)
 251         l->pending_plugins_tail = NULL;
 252       if (!gst_plugin_loader_spawn (l))
 253         return FALSE;
 254       break;
 255     }
 256   }
 257
 258   /* We exited after finding the crashing one. If there's any more pending,
 259    * dispatch them post-haste, but don't wait */
 260   for (cur = l->pending_plugins; cur != NULL; cur = next) {
 261     PendingPluginEntry *entry = (PendingPluginEntry *) (cur->data);
 262
 263     next = g_list_next (cur);
 264
 265     put_packet (l, PACKET_LOAD_PLUGIN, entry->tag,
 266         (guint8 *) entry->filename, strlen (entry->filename) + 1);
 267
 268     /* This might invalidate cur, which is why we grabbed 'next' above */
 269     if (!exchange_packets (l))
 270       goto restart;
 271   }
 272
 273   return TRUE;
 274 }
 275
 276 static gboolean
 277 plugin_loader_load_and_sync (GstPluginLoader * l, PendingPluginEntry * entry)
 278 {
 279   gint len;
 280
 281   GST_DEBUG_OBJECT (l->registry, "Synchronously loading plugin file %s",
 282       entry->filename);
 283
 284   len = strlen (entry->filename);
 285   put_packet (l, PACKET_LOAD_PLUGIN, entry->tag,
 286       (guint8 *) entry->filename, len + 1);
 287   put_packet (l, PACKET_SYNC, 0, NULL, 0);
 288
 289   l->rx_got_sync = FALSE;
 290   while (!l->rx_got_sync) {
 291     if (!exchange_packets (l))
 292       return FALSE;
 293   }
 294
 295   return TRUE;
 296 }
 297
 298 static void
 299 plugin_loader_create_blacklist_plugin (GstPluginLoader * l,
 300     PendingPluginEntry * entry)
 301 {
 302   GstPlugin *plugin = g_object_new (GST_TYPE_PLUGIN, NULL);
 303
 304   plugin->filename = g_strdup (entry->filename);
 305   plugin->file_mtime = entry->file_mtime;
 306   plugin->file_size = entry->file_size;
 307   plugin->flags |= GST_PLUGIN_FLAG_BLACKLISTED;
 308
 309   plugin->basename = g_path_get_basename (plugin->filename);
 310   plugin->desc.name = g_intern_string (plugin->basename);
 311   plugin->desc.description = g_strdup_printf ("Plugin for blacklisted file");
 312   plugin->desc.version = g_intern_string ("0.0.0");
 313   plugin->desc.license = g_intern_string ("BLACKLIST");
 314   plugin->desc.source = plugin->desc.license;
 315   plugin->desc.package = plugin->desc.license;
 316   plugin->desc.origin = plugin->desc.license;
 317
 318   GST_DEBUG ("Adding blacklist plugin '%s'", plugin->desc.name);
 319   gst_registry_add_plugin (l->registry, plugin);
 320 }
 321
 322 static gboolean
 323 gst_plugin_loader_spawn (GstPluginLoader * loader)
 324 {
 325   if (loader->child_running)
 326     return TRUE;
 327
 328   {
 329     /* FIXME: Find the plugin-scanner! */
 330     char *helper_bin =
 331         "/home/jan/devel/gstreamer/head/gstreamer/libs/gst/helpers/plugin-scanner";
 332     char *argv[] = { helper_bin, "-l", NULL };
 333
 334     if (!g_spawn_async_with_pipes (NULL, argv, NULL,
 335             G_SPAWN_DO_NOT_REAP_CHILD /* | G_SPAWN_STDERR_TO_DEV_NULL */ ,
 336             NULL, NULL, &loader->child_pid, &loader->fd_w.fd, &loader->fd_r.fd,
 337             NULL, NULL))
 338       return FALSE;
 339   }
 340
 341   gst_poll_add_fd (loader->fdset, &loader->fd_w);
 342   gst_poll_add_fd (loader->fdset, &loader->fd_r);
 343
 344   gst_poll_fd_ctl_read (loader->fdset, &loader->fd_r, TRUE);
 345
 346   loader->tx_buf_write = loader->tx_buf_read = 0;
 347
 348   loader->child_running = TRUE;
 349
 350   return TRUE;
 351 }
 352
 353 static void
 354 plugin_loader_cleanup_child (GstPluginLoader * l)
 355 {
 356   if (!l->child_running || l->is_child)
 357     return;
 358
 359   gst_poll_remove_fd (l->fdset, &l->fd_w);
 360   gst_poll_remove_fd (l->fdset, &l->fd_r);
 361
 362   close (l->fd_w.fd);
 363   close (l->fd_r.fd);
 364
 365 #ifndef G_OS_WIN32
 366   GST_LOG ("waiting for child process to exit");
 367   waitpid (l->child_pid, NULL, 0);
 368 #else
 369   g_warning ("FIXME: Implement child process shutdown for Win32");
 370 #endif
 371   g_spawn_close_pid (l->child_pid);
 372
 373   l->child_running = FALSE;
 374 }
 375
 376 gboolean
 377 _gst_plugin_loader_client_run ()
 378 {
 379   GstPluginLoader *l;
 380
 381   l = plugin_loader_new (NULL);
 382   if (l == NULL)
 383     return FALSE;
 384
 385   l->fd_w.fd = 1;               /* STDOUT */
 386   gst_poll_add_fd (l->fdset, &l->fd_w);
 387
 388   l->fd_r.fd = 0;               /* STDIN */
 389   gst_poll_add_fd (l->fdset, &l->fd_r);
 390   gst_poll_fd_ctl_read (l->fdset, &l->fd_r, TRUE);
 391
 392   l->is_child = TRUE;
 393
 394   GST_DEBUG ("Plugin scanner child running. Waiting for instructions");
 395
 396   /* Loop, listening for incoming packets on the fd and writing responses */
 397   while (!l->rx_done && exchange_packets (l));
 398
 399   plugin_loader_free (l);
 400
 401   return TRUE;
 402 }
 403
 404 static void
 405 put_packet (GstPluginLoader * l, guint type, guint32 tag,
 406     const guint8 * payload, guint32 payload_len)
 407 {
 408   guint8 *out;
 409   guint len = payload_len + HEADER_SIZE;
 410
 411   if (l->tx_buf_write + len >= l->tx_buf_size) {
 412     l->tx_buf_size = l->tx_buf_write + len + BUF_GROW_EXTRA;
 413     l->tx_buf = g_realloc (l->tx_buf, l->tx_buf_size);
 414   }
 415
 416   out = l->tx_buf + l->tx_buf_write;
 417
 418   out[0] = type;
 419   GST_WRITE_UINT24_BE (out + 1, tag);
 420   GST_WRITE_UINT32_BE (out + 4, payload_len);
 421   memcpy (out + HEADER_SIZE, payload, payload_len);
 422
 423   l->tx_buf_write += len;
 424   gst_poll_fd_ctl_write (l->fdset, &l->fd_w, TRUE);
 425 }
 426
 427 static void
 428 put_chunk (GstPluginLoader * l, GstRegistryChunk * chunk, guint * pos)
 429 {
 430   guint padsize = 0;
 431   guint len;
 432   guint8 *out;
 433
 434   /* Might need to align the chunk */
 435   if (chunk->align && ((*pos) % ALIGNMENT) != 0)
 436     padsize = ALIGNMENT - ((*pos) % ALIGNMENT);
 437
 438   len = padsize + chunk->size;
 439
 440   if (l->tx_buf_write + len >= l->tx_buf_size) {
 441     l->tx_buf_size = l->tx_buf_write + len + BUF_GROW_EXTRA;
 442     l->tx_buf = g_realloc (l->tx_buf, l->tx_buf_size);
 443   }
 444
 445   out = l->tx_buf + l->tx_buf_write;
 446   memcpy (out + padsize, chunk->data, chunk->size);
 447
 448   l->tx_buf_write += len;
 449   *pos += len;
 450
 451   gst_poll_fd_ctl_write (l->fdset, &l->fd_w, TRUE);
 452 };
 453
 454 static gboolean
 455 write_one (GstPluginLoader * l)
 456 {
 457   guint8 *out;
 458   guint32 to_write;
 459   int res;
 460
 461   if (l->tx_buf_read + HEADER_SIZE > l->tx_buf_write)
 462     return FALSE;
 463
 464   out = l->tx_buf + l->tx_buf_read;
 465   to_write = GST_READ_UINT32_BE (out + 4) + HEADER_SIZE;
 466   l->tx_buf_read += to_write;
 467
 468   GST_LOG ("Writing packet of size %d bytes to fd %d", to_write, l->fd_w.fd);
 469
 470   do {
 471     res = write (l->fd_w.fd, out, to_write);
 472     if (res > 0) {
 473       to_write -= res;
 474       out += res;
 475     }
 476   } while (to_write > 0 && res < 0 && (errno == EAGAIN || errno == EINTR));
 477   if (res < 0) {
 478     /* Failed to write -> child died */
 479     plugin_loader_cleanup_child (l);
 480     return FALSE;
 481   }
 482
 483   if (l->tx_buf_read == l->tx_buf_write) {
 484     gst_poll_fd_ctl_write (l->fdset, &l->fd_w, FALSE);
 485     l->tx_buf_read = l->tx_buf_write = 0;
 486   }
 487   return TRUE;
 488 }
 489
 490 static gboolean
 491 do_plugin_load (GstPluginLoader * l, const gchar * filename, guint tag)
 492 {
 493   GstPlugin *newplugin;
 494   GList *chunks = NULL;
 495
 496   GST_DEBUG ("Plugin scanner loading file %s. tag %u\n", filename, tag);
 497
 498 #if 0                           /* Test code - crash based on an env var */
 499   if (strstr (filename, "coreelements") == NULL) {
 500     g_printerr ("Crashing on file %s\n", filename);
 501     g_printerr ("%d", *(gint *) (NULL));
 502   }
 503 #endif
 504
 505
 506   newplugin = gst_plugin_load_file ((gchar *) filename, NULL);
 507   if (newplugin) {
 508     guint hdr_pos;
 509     guint offset;
 510
 511     /* Now serialise the plugin details and send */
 512     if (!_priv_gst_registry_chunks_save_plugin (&chunks,
 513             gst_registry_get_default (), newplugin))
 514       goto fail;
 515
 516     /* Store where the header is, write an empty one, then write
 517      * all the payload chunks, then fix up the header size */
 518     hdr_pos = l->tx_buf_write;
 519     offset = HEADER_SIZE;
 520     put_packet (l, PACKET_PLUGIN_DETAILS, tag, NULL, 0);
 521
 522     if (chunks) {
 523       GList *walk;
 524       for (walk = chunks; walk; walk = g_list_next (walk)) {
 525         GstRegistryChunk *cur = walk->data;
 526         put_chunk (l, cur, &offset);
 527
 528         if (!(cur->flags & GST_REGISTRY_CHUNK_FLAG_CONST))
 529           g_free (cur->data);
 530         g_free (cur);
 531       }
 532
 533       g_list_free (chunks);
 534
 535       /* Store the size of the written payload */
 536       GST_WRITE_UINT32_BE (l->tx_buf + hdr_pos + 4, offset - HEADER_SIZE);
 537     }
 538
 539     gst_object_unref (newplugin);
 540   } else {
 541     put_packet (l, PACKET_PLUGIN_DETAILS, tag, NULL, 0);
 542   }
 543
 544   return TRUE;
 545 fail:
 546   put_packet (l, PACKET_PLUGIN_DETAILS, tag, NULL, 0);
 547   if (chunks) {
 548     GList *walk;
 549     for (walk = chunks; walk; walk = g_list_next (walk)) {
 550       GstRegistryChunk *cur = walk->data;
 551
 552       if (!(cur->flags & GST_REGISTRY_CHUNK_FLAG_CONST))
 553         g_free (cur->data);
 554       g_free (cur);
 555     }
 556
 557     g_list_free (chunks);
 558   }
 559
 560   return FALSE;
 561 }
 562
 563 static gboolean
 564 handle_rx_packet (GstPluginLoader * l,
 565     guint pack_type, guint32 tag, guint8 * payload, guint payload_len)
 566 {
 567   gboolean res = TRUE;
 568
 569   switch (pack_type) {
 570     case PACKET_EXIT:
 571       gst_poll_fd_ctl_read (l->fdset, &l->fd_r, FALSE);
 572       if (l->is_child) {
 573         /* Respond, then we keep looping until the parent closes the fd */
 574         put_packet (l, PACKET_EXIT, 0, NULL, 0);
 575       } else {
 576         l->rx_done = TRUE;      /* All done reading from child */
 577       }
 578       return TRUE;
 579     case PACKET_LOAD_PLUGIN:{
 580       if (!l->is_child)
 581         return TRUE;
 582
 583       /* Payload is the filename to load */
 584       res = do_plugin_load (l, (gchar *) payload, tag);
 585
 586       break;
 587     }
 588     case PACKET_PLUGIN_DETAILS:{
 589       gchar *tmp = (gchar *) payload;
 590       PendingPluginEntry *entry = NULL;
 591       GList *cur;
 592
 593       GST_DEBUG_OBJECT (l->registry,
 594           "Received plugin details from child w/ tag %u. %d bytes info",
 595           tag, payload_len);
 596
 597       /* Assume that tagged details come back in the order
 598        * we requested, and delete anything before this one */
 599       cur = l->pending_plugins;
 600       while (cur) {
 601         PendingPluginEntry *e = (PendingPluginEntry *) (cur->data);
 602
 603         if (e->tag > tag)
 604           break;
 605
 606         cur = g_list_delete_link (cur, cur);
 607
 608         if (e->tag == tag) {
 609           entry = e;
 610           break;
 611         } else {
 612           g_free (e->filename);
 613           g_free (e);
 614         }
 615       }
 616       l->pending_plugins = cur;
 617       if (cur == NULL)
 618         l->pending_plugins_tail = NULL;
 619
 620       if (payload_len > 0) {
 621         GstPlugin *newplugin;
 622         _priv_gst_registry_chunks_load_plugin (l->registry, &tmp,
 623             tmp + payload_len, &newplugin);
 624         newplugin->flags &= ~GST_PLUGIN_FLAG_CACHED;
 625         GST_LOG_OBJECT (l->registry,
 626             "marking plugin %p as registered as %s", newplugin,
 627             newplugin->filename);
 628         newplugin->registered = TRUE;
 629
 630         /* We got a set of plugin details - remember it for later */
 631         l->got_plugin_details = TRUE;
 632       } else if (entry != NULL) {
 633         /* Create a blacklist entry for this file to prevent scanning every time */
 634         plugin_loader_create_blacklist_plugin (l, entry);
 635         l->got_plugin_details = TRUE;
 636       }
 637
 638       if (entry != NULL) {
 639         g_free (entry->filename);
 640         g_free (entry);
 641       }
 642
 643       break;
 644     }
 645     case PACKET_SYNC:
 646       if (l->is_child) {
 647         /* Respond with our reply - also a sync */
 648         put_packet (l, PACKET_SYNC, 0, NULL, 0);
 649         GST_LOG ("Got SYNC in child - replying");
 650       } else
 651         l->rx_got_sync = TRUE;
 652       break;
 653     default:
 654       return FALSE;             /* Invalid packet -> something is wrong */
 655   }
 656
 657   return res;
 658 }
 659
 660 static gboolean
 661 read_one (GstPluginLoader * l)
 662 {
 663   guint32 to_read, packet_len, tag;
 664   guint8 *in;
 665   gint res;
 666
 667   to_read = HEADER_SIZE;
 668   in = l->rx_buf;
 669   do {
 670     res = read (l->fd_r.fd, in, to_read);
 671     if (res > 0) {
 672       to_read -= res;
 673       in += res;
 674     }
 675   } while (to_read > 0 && res < 0 && (errno == EAGAIN || errno == EINTR));
 676
 677   if (res < 0) {
 678     GST_LOG ("Failed reading packet header");
 679     return FALSE;
 680   }
 681
 682   packet_len = GST_READ_UINT32_BE (l->rx_buf + 4);
 683
 684   if (packet_len + HEADER_SIZE >= l->rx_buf_size) {
 685     l->rx_buf_size = packet_len + HEADER_SIZE + BUF_GROW_EXTRA;
 686     l->rx_buf = g_realloc (l->rx_buf, l->rx_buf_size);
 687   }
 688
 689   in = l->rx_buf + HEADER_SIZE;
 690   to_read = packet_len;
 691   do {
 692     res = read (l->fd_r.fd, in, to_read);
 693     if (res > 0) {
 694       to_read -= res;
 695       in += res;
 696     }
 697   } while (to_read > 0 && res < 0 && (errno == EAGAIN || errno == EINTR));
 698
 699   if (res < 0) {
 700     GST_ERROR ("Packet payload read failed");
 701     return FALSE;
 702   }
 703
 704   tag = GST_READ_UINT24_BE (l->rx_buf + 1);
 705
 706   return handle_rx_packet (l, l->rx_buf[0], tag,
 707       l->rx_buf + HEADER_SIZE, packet_len);
 708 }
 709
 710 static gboolean
 711 exchange_packets (GstPluginLoader * l)
 712 {
 713   gint res;
 714
 715   /* Wait for activity on our FDs */
 716   do {
 717     do {
 718       res = gst_poll_wait (l->fdset, GST_CLOCK_TIME_NONE);
 719     } while (res == -1 && (errno == EINTR || errno == EAGAIN));
 720
 721     if (res < 0)
 722       return FALSE;
 723
 724     GST_LOG ("Poll res = %d. %d bytes pending for write", res,
 725         l->tx_buf_write - l->tx_buf_read);
 726
 727     if (!l->rx_done) {
 728       if (gst_poll_fd_has_error (l->fdset, &l->fd_r) ||
 729           gst_poll_fd_has_closed (l->fdset, &l->fd_r)) {
 730         GST_LOG ("read fd %d closed/errored", l->fd_r.fd);
 731         plugin_loader_cleanup_child (l);
 732         return FALSE;
 733       }
 734
 735       if (gst_poll_fd_can_read (l->fdset, &l->fd_r)) {
 736         if (!read_one (l))
 737           return FALSE;
 738       }
 739     }
 740
 741     if (l->tx_buf_read < l->tx_buf_write) {
 742       if (gst_poll_fd_has_error (l->fdset, &l->fd_w) ||
 743           gst_poll_fd_has_closed (l->fdset, &l->fd_r)) {
 744         GST_ERROR ("write fd %d closed/errored", l->fd_w.fd);
 745         plugin_loader_cleanup_child (l);
 746         return FALSE;
 747       }
 748       if (gst_poll_fd_can_write (l->fdset, &l->fd_w)) {
 749         if (!write_one (l))
 750           return FALSE;
 751       }
 752     }
 753   } while (l->tx_buf_read < l->tx_buf_write);
 754
 755   return TRUE;
 756 }