workshop: add "stdin" column

Allows passing arbitrary (binary) data to a job's stdin.  For
simplicity, this is implemented using a memfd instead of a pair of
pipes.

Author: MaxKellermann

debian/changelog

@@ -2,6 +2,7 @@ cm4all-workshop (7.1) unstable; urgency=low
 
   * workshop: disable progress pipe if "translate" is enabled
   * workshop: read remaining output after job process exits
+  * workshop: add "stdin" column
 
  --   
 

doc/index.rst

@@ -536,6 +536,8 @@ The `jobs` table
 * ``env``: Additional environment variables.  Some dangerous
   environment variables cannot be set, though, for example
   ``LD_PRELOAD``.
+* ``stdin``: If not ``NULL``, then the process is started with a file
+  handle on standard input that reads this data.
 * ``node_name``: Name of the node which is currently executing
   this job, or :samp:`NULL`.
 * ``node_timeout``: When this time stamp has passed, then the

sql/jobs.sql

@@ -29,6 +29,8 @@ CREATE TABLE jobs (
         args varchar(4096)[] NULL,
         -- environment variables in the form NAME=VALUE
         env varchar(4096)[] NULL,
+        -- optional data fed into stdin
+        stdin bytea NULL,
 
         -- which cm4all-workshop node is executing this job?
         node_name varchar(256) NULL,

src/Migrate.cxx

@@ -47,6 +47,9 @@ MigrateWorkshopDatabase(Pg::Connection &c, const char *schema)
 
 	/* since Workshop 6.0.1 */
 	c.Execute("ALTER TABLE jobs ADD COLUMN IF NOT EXISTS cpu_usage interval NULL");
+
+	/* since Workshop 7.1 */
+	c.Execute("ALTER TABLE jobs ADD COLUMN IF NOT EXISTS stdin bytea NULL");
 }
 
 static void

src/workshop/Job.hxx

@@ -4,7 +4,10 @@
 
 #pragma once
 
+#include "util/AllocatedArray.hxx"
+
 #include <chrono>
+#include <cstddef>
 #include <string>
 #include <forward_list>
 
@@ -19,6 +22,8 @@ struct WorkshopJob {
 
 	std::forward_list<std::string> env;
 
+	AllocatedArray<std::byte> stdin;
+
 	explicit WorkshopJob(WorkshopQueue &_queue):queue(_queue) {}
 
 	WorkshopJob(WorkshopQueue &_queue, const char *_id, const char *_plan_name)

src/workshop/Operator.cxx

@@ -25,6 +25,7 @@
 #include "net/EasyMessage.hxx"
 #include "net/SocketPair.hxx"
 #include "net/UniqueSocketDescriptor.hxx"
+#include "io/linux/MemFD.hxx"
 #include "io/FdHolder.hxx"
 #include "io/FileAt.hxx"
 #include "io/Open.hxx"
@@ -259,6 +260,9 @@ WorkshopOperator::Start2(std::size_t max_log_buffer,
 			    stderr_w, control_child,
 			    close_fds);
 
+	if (job.stdin != nullptr)
+		p.stdin_fd = close_fds.Insert(CreateMemFD("stdin", job.stdin));
+
 	/* use a per-plan cgroup */
 
 	CgroupOptions cgroup;

src/workshop/PGQueue.cxx

@@ -86,7 +86,7 @@ pg_select_new_jobs(Pg::Connection &db,
 	assert(plans_exclude != nullptr && *plans_exclude == '{');
 	assert(plans_lowprio != nullptr && *plans_lowprio == '{');
 
-	const char *sql = "SELECT id,plan_name,args,env "
+	const char *sql = "SELECT id,plan_name,args,env,stdin "
 		"FROM jobs "
 		"WHERE node_name IS NULL "
 		"AND time_done IS NULL AND exit_status IS NULL "

src/workshop/Queue.cxx

@@ -8,6 +8,7 @@
 #include "Plan.hxx"
 #include "lib/fmt/RuntimeError.hxx"
 #include "pg/Array.hxx"
+#include "pg/Hex.hxx"
 #include "pg/Reflection.hxx"
 #include "event/Loop.hxx"
 #include "util/StringAPI.hxx"
@@ -88,6 +89,9 @@ MakeJob(WorkshopQueue &queue,
 	job.args = Pg::DecodeArray(result.GetValue(row, 2));
 	job.env = Pg::DecodeArray(result.GetValue(row, 3));
 
+	if (!result.IsValueNull(row, 4))
+		job.stdin = Pg::DecodeHex(result.GetValueView(row, 4));
+
 	if (job.id.empty())
 		throw std::runtime_error("Job has no id");
 
@@ -459,6 +463,7 @@ WorkshopQueue::OnConnect()
 	static constexpr const char *const required_jobs_columns[] = {
 		"enabled",
 		"log",
+		"stdin",
 	};
 
 	const char *schema = db.GetSchemaName().empty()