- int ret, i, need_more_writes = 1;
- size_t min_written = 0;
-
- while (need_more_writes) {
- need_more_writes = 0;
- FOR_EACH_WRITER_NODE(i, g) {
- size_t w = g->written[i];
- unsigned char *p = buf + w;
- int bytes_to_write;
- struct writer_node *wn = &g->writer_nodes[i];
- if (!i)
- min_written = w;
- else
- min_written = PARA_MIN(min_written, w);
- if (w == *loaded)
- continue;
- if (!g->eof && (*loaded < wn->chunk_bytes + w))
- continue;
- bytes_to_write = PARA_MIN(wn->chunk_bytes,
- *loaded - w);
- ret = wn->writer->write(p, bytes_to_write, wn);
- if (ret < 0)
- goto out;
- if (ret != bytes_to_write)
- PARA_WARNING_LOG("short write: %d/%d\n", ret,
- bytes_to_write);
- g->written[i] += ret;
- need_more_writes = 1;
- }
+ struct writer_node_group *g = container_of(t, struct writer_node_group, task);
+ int i;
+
+ FOR_EACH_WRITER_NODE(i, g) {
+ struct writer_node *wn = &g->writer_nodes[i];
+ struct writer *w = writers + wn->writer_num;
+ if (!w->pre_select)
+ continue;
+ t->error = w->pre_select(s, wn);
+ if (t->error < 0)
+ return;
+ }
+ /*
+ * Force a minimal delay if something was written during the previous
+ * call to wng_post_select(). This is necessary because the filter
+ * chain might still have data for us which it couldn't convert during
+ * the previous run due to its buffer size constraints. In this case we
+ * do not want to wait until the next input data arrives as this could
+ * lead to buffer underruns.
+ */
+ if (g->last_written == 0)
+ return;
+ s->timeout.tv_sec = 0;
+ s->timeout.tv_usec = 1;
+}
+
+static void wng_post_select(struct sched *s, struct task *t)
+{
+ struct writer_node_group *g = container_of(t, struct writer_node_group, task);
+ int i;
+ size_t min_written = 0, max_written = 0;
+
+ FOR_EACH_WRITER_NODE(i, g) {
+ struct writer_node *wn = &g->writer_nodes[i];
+ struct writer *w = writers + wn->writer_num;
+ t->error = w->post_select(s, wn);
+ if (t->error < 0)
+ return;
+ if (!i)
+ min_written = wn->written;
+ else
+ min_written = PARA_MIN(min_written, wn->written);
+ max_written = PARA_MAX(max_written, wn->written);
+ }
+ g->last_written = max_written;
+ //PARA_INFO_LOG("loaded: %zd, min_written: %zd bytes\n", *g->loaded, min_written);
+ if (min_written) {
+ *g->loaded -= min_written;
+ FOR_EACH_WRITER_NODE(i, g)
+ g->writer_nodes[i].written -= min_written;
+ }
+ if (!*g->loaded && *g->input_error) {
+ t->error = *g->input_error;
+ return;
+ }
+ if (*g->loaded && min_written) {
+// PARA_INFO_LOG("moving %zd bytes\n", *g->loaded);
+ memmove(*g->bufp, *g->bufp + min_written, *g->loaded);