Light task-processing systems with zero dependency for Linux systems

Question

I made a prototype for a task processing systems.

Architecture

• Worker - is a entity that is processing one-by-one tasks for the same processor.
• Processor - (also: executable, app) executable file (language agnostic) that getting input from STDIN and returning result to STDOUT with optional logs to STDERR. Return code 0 means success, otherwise - fail.
• Task - single processing unit that contains saved file as STDIN for the processor

Briefly - worker gets file from tasks directory, feeds it to STDIN of executable.

On success (ret code 0) - saves STDOUT to result dir and remove task file.

On fail - moves task file to requeue directory. It will be later moved back to tasks directory

Logic is quite straight forward and could be described by one diagram

Task states

I would like to kindly ask community to review the github code. Please advise me how to make better/faster/more idiomatic (C Linux style). I will very appreciate it.

worker.h

/**
 * Definition of worker in a directory.
 * Single directory should be handled by single worker
 */
typedef struct worker_t {
  char *root;         // root directory
  char *name;         // worker ID/name - base name of root directory
  char *tasks_dir;    // directory with files that will be fed to the application as STDIN (tasks)
  char *requeue_dir;  // directory where failed tasks will be added for later re-processing
  char *progress_dir; // directory where stdout of application will be stored and used as result (on success)
  char *complete_dir; // directory where stdout of successful tasks execution will be stored
  char *bin_dir;      // directory where application should be stored (first executable file)
} worker_t;

/**
 * Initialize worker. Function uses malloc internally to allocate and set worker structure.
 * Allocated worker should be destroyed by `worker_destroy`
 * @param worker pointer to worker
 * @param root_dir root directory that will be used as parent for all nested dirs (tasks, progress, ..)
 */
void worker_init(worker_t *worker, const char *root_dir);

/**
 * Free all allocated resource in worker
 * @param worker pre-initialized by `worker_init` instance
 */
void worker_destroy(worker_t *worker);

/**
 * Prepare environment for worker: creates directories (however, root directory should be already created),
 * clean old files and so on.
 * @param worker initialized instance of worker
 * @return 0 on success, otherwise print error to stderr and return related code
 */
int worker_prepare(const worker_t *worker);

/**
 * Run all queue tasks: sequentially execute `worker_run_one` for each file in tasks dir, sorted alphabetically.
 * Fail-tolerant: print error if tasks failed but do not stop next executions.
 * Mostly error could return if no executable found in bin-dir
 * @param worker  initialized instance of worker
 * @return 0 on success, otherwise print error to stderr and return related code
 */
int worker_run_all(const worker_t *worker);

/**
 * Execute only one task identified by name (base name of task file in tasks dir).
 * In case of failure - task will be automatically re-queued, in case of successful (return code 0) execution -
 * stdout will be stored in complete-dir with same name as task.
 * @param worker initialized instance of worker
 * @param task_name name of task  (base name of task file in tasks dir)
 * @return 0 on success, otherwise print error to stderr and return related code
 */
int worker_run_one(const worker_t *worker, const char *task_name);

/**
 * Initialize inotify subsystem to watch tasks directory.
 * @param worker initialized instance of worker
 * @param listener reference to where inotify file descriptor will be put
 * @return 0 on success, otherwise print error to stderr and return related code
 */
int worker_create_listener(const worker_t *worker, int *listener);

/**
 * Wait or inotify events
 * @param worker initialized instance of worker
 * @param listener inotify file descriptor (commonly from `worker_create_listener`)
 * @return 0 on success, otherwise print error to stderr and return related code
 */
int worker_listen(const worker_t *worker, int listener);

/**
 * Check files in re-queue directory for expired tasks and move them back to tasks dir
 * @param worker initialized instance of worker
 * @param expiration_sec expiration time in seconds
 * @return 0 on success, otherwise print error to stderr and return related code
 */
int worker_requeue_check(const worker_t *worker, __time_t expiration_sec);

worker.c

#include <dirent.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/inotify.h>
#include <sys/time.h>
#include <unistd.h>
#include <errno.h>
#include <stdlib.h>
#include "task.h"
#include "utils.h"

void worker_init(worker_t *worker, const char *root_dir) {
  char *bin_dir = filepath_join(root_dir, "app");
  worker->root = strdup(root_dir);
  worker->name = strdup(filepath_basename(root_dir));
  worker->bin_dir = bin_dir;
  worker->progress_dir = filepath_join(worker->root, "progress");
  worker->requeue_dir = filepath_join(worker->root, "requeue");
  worker->tasks_dir = filepath_join(worker->root, "tasks");
  worker->complete_dir = filepath_join(worker->root, "complete");
}

void worker_destroy(worker_t *worker) {
  if (!worker) {
    return;
  }
  free(worker->root);
  free(worker->name);
  free(worker->progress_dir);
  free(worker->requeue_dir);
  free(worker->tasks_dir);
  free(worker->complete_dir);
  free(worker->bin_dir);
}

static const mode_t dir_mode = S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH;

int worker_create_dirs(const worker_t *worker) {
  int ret;
  ret = mkdir(worker->tasks_dir, dir_mode);
  if (ret != 0 && errno != EEXIST) {
    perror("retask: create task dir");
    return ret;
  }
  ret = mkdir(worker->progress_dir, dir_mode);
  if (ret != 0 && errno != EEXIST) {
    perror("retask: create progress dir");
    return ret;
  }
  ret = mkdir(worker->requeue_dir, dir_mode);
  if (ret != 0 && errno != EEXIST) {
    perror("retask: create requeue dir");
    return ret;
  }
  ret = mkdir(worker->complete_dir, dir_mode);
  if (ret != 0 && errno != EEXIST) {
    perror("retask: create complete dir");
    return ret;
  }
  ret = mkdir(worker->bin_dir, dir_mode);
  if (ret != 0 && errno != EEXIST) {
    perror("retask: create bin dir");
    return ret;
  }
  return 0;
}

int worker_clean_progress(const worker_t *worker) {
  DIR *dp;
  struct dirent *ep;
  dp = opendir(worker->progress_dir);
  if (!dp) {
    fprintf(stderr, "retask: open progress dir %s: %s\n", worker->progress_dir, strerror(errno));
    return errno;
  }
  while ((ep = readdir(dp))) {
    if (ep->d_type == DT_DIR) {
      continue;
    }
    char *path = filepath_join(worker->progress_dir, ep->d_name);
    int ret = remove(path);
    if (ret != 0) {
      fprintf(stderr, "retask: remove progress file %s: %s\n", path, strerror(errno));
    }
    free(path);
  }
  closedir(dp);
  return 0;
}

int worker_prepare(const worker_t *worker) {
  int ret;
  ret = worker_create_dirs(worker);
  if (ret != 0) {
    return ret;
  }
  ret = worker_clean_progress(worker);
  if (ret != 0) {
    return ret;
  }
  return 0;
}

int worker_run_all(const worker_t *worker) {
  struct dirent **namelist;

  int n = scandir(worker->tasks_dir, &namelist, NULL, alphasort);
  if (n == -1) {
    perror("retask: scandir for tasks");
    return -1;
  }

  for (int i = 0; i < n; ++i) {
    struct dirent *entry = namelist[i];
    if (entry->d_type == DT_DIR) {
      continue;
    }
    int ret = worker_run_one(worker, entry->d_name);
    if (ret != 0) {
      fprintf(stderr, "retask: task %s failed with code %i\n", entry->d_name, ret);
    }
    free(entry);
  }
  free(namelist);
  return 0;
}

int worker_run_one(const worker_t *worker, const char *task_name) {
  task_t task;
  int ret = task_init(&task, worker, task_name);
  if (ret != 0) {
    return ret;
  }
  ret = task_run(&task);
  task_destroy(&task);
  fprintf(stderr, "retask: task %s executed with code %i\n", task_name, ret);
  return ret;
}

int worker_create_listener(const worker_t *worker, int *listener) {
  int fd = inotify_init();
  if (fd == -1) {
    fprintf(stderr, "retask: failed to add watcher on directory %s: %s\n", worker->tasks_dir, strerror(errno));
    return -1;
  }
  int rc = inotify_add_watch(fd, worker->tasks_dir, IN_MOVED_TO | IN_CLOSE_WRITE);
  if (rc == -1) {
    fprintf(stderr, "retask: failed to add watcher on directory %s: %s\n", worker->tasks_dir, strerror(errno));
    close(fd);
    return -1;
  }
  *listener = fd;
  return 0;
}

int worker_listen(const worker_t *worker, int listener) {
  const struct inotify_event *event;
  char *ptr;
  char buf[4096] __attribute__ ((aligned(__alignof__(struct inotify_event))));
  ssize_t len = read(listener, buf, sizeof(buf));
  if (len == -1 && errno != EAGAIN) {
    fprintf(stderr, "retask: failed to listen events on directory %s: %s\n", worker->tasks_dir, strerror(errno));
    return -1;
  }
  for (ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
    event = (const struct inotify_event *) ptr;
    if (event->len)
      fprintf(stderr, "retask: detected %s", event->name);
    if (event->mask & IN_ISDIR)
      fprintf(stderr, " [directory]\n");
    else
      fprintf(stderr, " [file]\n");
  }

  return 0;
}

int worker_requeue_check(const worker_t *worker, __time_t expiration_sec) {
  DIR *dp;
  struct dirent *ep;
  struct stat info;
  struct timeval now;
  int ret = gettimeofday(&now, NULL);
  if (ret != 0) {
    fprintf(stderr, "retask: get current date/time: %s\n", strerror(errno));
    return ret;
  }
  dp = opendir(worker->requeue_dir);
  if (!dp) {
    ret = errno;
    fprintf(stderr, "retask: open requeue dir %s: %s\n", worker->requeue_dir, strerror(errno));
    return ret;
  }
  while ((ep = readdir(dp))) {
    if (ep->d_type == DT_DIR) {
      continue;
    }
    char *src = filepath_join(worker->requeue_dir, ep->d_name);

    ret = stat(src, &info);
    if (ret != 0) {
      fprintf(stderr, "retask: stat queued file %s: %s\n", src, strerror(errno));
      free(src);
      continue;
    }
    if (info.st_ctim.tv_sec + expiration_sec > now.tv_sec) {
      continue;
    }
    char *dst = filepath_join(worker->tasks_dir, ep->d_name);
    ret = rename(src, dst);
    if (ret != 0) {
      fprintf(stderr, "retask: failed move queued file %s to tasks %s: %s\n", src, dst, strerror(errno));
    } else {
      fprintf(stderr, "retask: re-queued file %s due to expiration time\n", src);
    }
    free(dst);
    free(src);
  }
  closedir(dp);
  return 0;
}

task.h

#include "worker.h"

/**
 * Definition of single task inside worker. Worker should not be destroyed before task.
 */
typedef struct task_t {
  char *name;             // task name (basename of task file)
  char *file;             // task file. The file will be used as STDIN for the executable
  char *progress_file;    // temp file for STDOUT of the executable
  char *result_file;      // destination of progress file in case of successful execution (ret code 0)
  char *requeue_file;     // destination for task file in case of failed execution (for re-queue)
  char *executable;       // executable (first file marked as executable inside bin dir)
  const worker_t *worker; // reference to parent worker instance
} task_t;

/**
 * Initialize internal resource for task, allocate paths, detect executable in a bin dir.
 * Paths will be allocated by malloc.
 * @param task reference to task that will be initialized
 * @param worker initialized instance of worker
 * @param task_name task name (basename of file in tasks dir)
 * @return 0 on success, otherwise print error to stderr and return related code
 */
int task_init(task_t *task, const worker_t *worker, const char *task_name);

/**
 * Destroy (free) allocated resource (worker ref will not be touched)
 * @param task initialized instance of task
 */
void task_destroy(task_t *task);

/**
 * Run executable and process result files (re-queue or move to complete). This function internally runs fork and fill
 * following environment variables: WORKER_ID, WORKER_ROOT_DIR, WORKER_BIN_DIR, WORKER_TASKS_DIR, TASK_ID, TASK_EXECUTABLE.
 * Task file will be used as STDIN and progress file as STDOUT. STDERR will be mapped as parent process.
 * @param task initialized instance of task
 * @return  0 on success, otherwise print error to stderr and return related code
 */
int task_run(const task_t *task);

task.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/wait.h>
#include <sys/prctl.h>
#include "../include/utils.h"
#include "../include/task.h"

int __run_app(const task_t *task, int stdin_file, int stdout_file);

int task_init(task_t *task, const worker_t *worker, const char *task_name) {
  char *executable = find_executable(worker->bin_dir);
  if (!executable) {
    return -1;
  }
  task->name = strdup(task_name);
  task->file = filepath_join(worker->tasks_dir, task_name);
  task->progress_file = filepath_join(worker->progress_dir, task_name);
  task->result_file = filepath_join(worker->complete_dir, task_name);
  task->requeue_file = filepath_join(worker->requeue_dir, task_name);
  task->executable = executable;
  task->worker = worker;
  return 0;
}

void task_destroy(task_t *task) {
  free(task->file);
  free(task->progress_file);
  free(task->result_file);
  free(task->requeue_file);
  free(task->executable);
  free(task->name);
}

int task_run_app(const task_t *task) {
  FILE *progress = fopen(task->progress_file, "w");
  if (!progress) {
    fprintf(stderr, "retask: create progress file %s: %s", task->progress_file, strerror(errno));
    return 1;
  }
  FILE *source = fopen(task->file, "r");
  if (!source) {
    fprintf(stderr, "retask: read task file %s: %s", task->file, strerror(errno));
    fclose(progress);
    return 2;
  }

  int ret = __run_app(task, fileno(source), fileno(progress));
  fclose(source);
  fclose(progress);
  return ret;
}

int __run_app(const task_t *task, int stdin_file, int stdout_file) {
  pid_t pid = 0;
  pid = fork();
  if (pid == 0) {
    dup2(stdin_file, STDIN_FILENO);
    dup2(stdout_file, STDOUT_FILENO);
    prctl(PR_SET_PDEATHSIG, SIGTERM);

    setenv("WORKER_ID", task->worker->name, 1);
    setenv("WORKER_ROOT_DIR", task->worker->root, 1);
    setenv("WORKER_BIN_DIR", task->worker->bin_dir, 1);
    setenv("WORKER_TASKS_DIR", task->worker->tasks_dir, 1);
    setenv("TASK_ID", task->name, 1);
    setenv("TASK_EXECUTABLE", task->executable, 1);

    execl(task->executable, task->executable, (char *) NULL);
    _exit(1);
  }
  int status;
  if (waitpid(pid, &status, 0) == -1) {
    perror("waitpid() failed");
    return 1;
  }
  if (WIFEXITED(status)) {
    return WEXITSTATUS(status);
  }
  return status;
}

int task_complete(const task_t *task, int ret_code) {
  if (ret_code != 0) {
    remove(task->progress_file);
    return rename(task->file, task->requeue_file);
  }
  ret_code = rename(task->progress_file, task->result_file);
  if (ret_code != 0) {
    return ret_code;
  }
  return remove(task->file);
}

int task_run(const task_t *task) {
  int ret_code = task_run_app(task);
  ret_code = task_complete(task, ret_code);

  return ret_code;
}

utils.h

/**
 * Concat root dir and child with respect of root dir ending (with / or without).
 * It is caller responsible to free returned memory.
 * @param root_dir parent directory
 * @param child child directory
 * @return joined path
 */
char *filepath_join(const char *root_dir, const char *child);

/**
 * Find first file with executable flag.
 * It is caller responsible to free returned memory.
 * @param bin_dir directory to scan
 * @return full path on success, otherwise print error to stderr and return NULL
 */
char *find_executable(const char *bin_dir);

/**
 * Get basename (last part after /). There is no memory allocation here.
 * @param root_dir path to scan
 * @return ref to the first symbol in root_dir var
 */
const char *filepath_basename(const char *root_dir);

utils.c

#include <dirent.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <errno.h>
#include "../include/utils.h"

char *find_executable(const char *bin_dir) {
  DIR *dirp;
  struct dirent *direntp;
  dirp = opendir(bin_dir);
  if (dirp == NULL) {
    fprintf(stderr, "can't detect executable in %s: %s\n", bin_dir, strerror(errno));
    return NULL;
  }
  for (;;) {
    direntp = readdir(dirp);
    if (direntp == NULL) {
      break;
    }
    if (direntp->d_type == DT_DIR) {
      continue;
    }
    char *path = filepath_join(bin_dir, direntp->d_name);
    if (access(path, X_OK) != -1) {
      closedir(dirp);
      return path;
    }
    free(path);
  }
  closedir(dirp);
  fprintf(stderr, "there is no executable file in %s\n", bin_dir);
  return NULL;
}

char *filepath_join(const char *root_dir, const char *child) {
  size_t child_len = strlen(child);
  size_t root_len = strlen(root_dir);
  size_t n = root_len + child_len + 1;
  char need_slash = 0;
  if (root_len > 0 && root_dir[root_len - 1] != '/') {
    ++n;
    need_slash = 1;
  }
  char *data = (char *) malloc(n);
  memcpy(data, root_dir, root_len);
  if (need_slash) {
    data[root_len] = '/';
  }
  memcpy(&data[root_len + need_slash], child, child_len);
  data[n - 1] = '\0';
  return data;
}

const char *filepath_basename(const char *root_dir) {
  size_t len = strlen(root_dir);
  if (len == 0) {
    return root_dir;
  }
  for (ssize_t i = len - 1; i >= 0; --i) {
    if (root_dir[i] == '/') {
      return &root_dir[i + 1];
    }
  }
  return root_dir;
}

main.c

#include <unistd.h>
#include <stdlib.h>
#include <sys/prctl.h>
#include <stdio.h>
#include <wait.h>
#include <string.h>
#include <errno.h>
#include <stdnoreturn.h>
#include "worker.h"

static const long DEFAULT_REQUEUE_INTERVAL = 5;
static const char DEFAULT_LOCATION[] = ".";
static const char REQUEUE_PROCESS[] = "requeue";

noreturn void run_requeue(worker_t *worker, long requeue_interval) {
  long wait_seconds = requeue_interval / 2;
  if (wait_seconds < 1) {
    wait_seconds = 1;
  }
  for (;;) {
    int ret = worker_requeue_check(worker, requeue_interval);
    if (ret != 0) {
      fprintf(stderr, "retask: requeue failed with code %i\n", ret);
    }
    sleep(wait_seconds);
  }
}

int run_worker(worker_t *worker) {
  int ret;
  int listener;
  ret = worker_create_listener(worker, &listener);
  if (ret != 0) {
    return ret;
  }
  for (;;) {
    ret = worker_run_all(worker);
    if (ret != 0) {
      break;
    }
    ret = worker_listen(worker, listener);
    if (ret != 0) {
      break;
    }
  }
  close(listener);
  return 0;
}

int launch(worker_t *worker, long requeue_interval) {
  int ret = worker_prepare(worker);
  if (ret != 0) {
    return ret;
  }

  if (fork() == 0) {
    // sub process to manage requeue process
    size_t name_len = strlen(worker->name);
    char *process_name = (char *) malloc(name_len + 1 + sizeof(REQUEUE_PROCESS));
    memcpy(process_name, worker->name, name_len);
    process_name[name_len] = '/';
    memcpy(&process_name[name_len + 1], REQUEUE_PROCESS, sizeof(REQUEUE_PROCESS));
    prctl(PR_SET_NAME, process_name, 0, 0, 0);
    prctl(PR_SET_PDEATHSIG, SIGTERM);
    free(process_name);
    run_requeue(worker, requeue_interval);
    _exit(EXIT_FAILURE);
  }

  return run_worker(worker);
}

void usage() {
  fprintf(stderr, "Executor for tasks from filesystem\n");
  fprintf(stderr, "\n");
  fprintf(stderr, "Author: Baryshnikov Aleksandr <[email protected]>\n");
  fprintf(stderr, "\n");
  fprintf(stderr, "retask [flags]\n");
  fprintf(stderr, "    -c <directory> [default=.]  use specified directory as work dir\n");
  fprintf(stderr, "    -r <seconds>   [default=%li]  interval for requeue\n", DEFAULT_REQUEUE_INTERVAL);
}

int main(int argc, char *argv[]) {
  int opt;
  const char *location = DEFAULT_LOCATION;
  long requeue_sec = DEFAULT_REQUEUE_INTERVAL;

  while ((opt = getopt(argc, argv, "c:r:")) != -1) {
    switch (opt) {
    case 'r': {
      requeue_sec = strtol(optarg, NULL, 10);
      if (errno == EINVAL || errno == ERANGE) {
        fprintf(stderr, "invalid value for requeue interval: %s\n", optarg);
        usage();
        exit(EXIT_FAILURE);
      }
      break;
    }
    case 'c': {
      location = optarg;
      break;
    }
    default: /* '?' */
      usage();
      exit(EXIT_FAILURE);
    }
  }

  worker_t worker;
  worker_init(&worker, location);
  int ret = launch(&worker, requeue_sec);
  worker_destroy(&worker);
  return ret;
}

Answer 1

It's a good idea for worker.c to include "worker.h", preferably as the very first line. That ensures that the declarations are consistent and that the header has no missing includes of its own. (So, in task.c move the task.h include to the top spot, and similarly with utils.h in utils.c.)

I recommend adding include-guards on all your header files. I know there's no intention to multiply-include any of them, but it can easily happen, particularly when they include each other.

Your error-checking on I/O operations is commendable - well done. However, I noticed an unchecked malloc() in filepath_join() that should be addressed (in passing, it also has a pointless cast). All the callers of filepath_join() need to be prepared for that, too.

There's a similar unchecked malloc() and pointless cast in main(), where we initialise process_name.

In main(), we shouldn't assume that errno is unset when we call strtol() - if it's successful, then errno may be left unchanged.

Answer 2

Make sure Doxygen picks up comments after struct members

It's good to add a comment after each member in a struct declaration to describe what that member does. To make Doxygen pick up the member documentation as well, use //!< or ///< instead of // to start the comments with.

Use "filename" or "path", not "file" describe filenames

You will be dealing with both file handles (either a FILE * or something like int fd) as well as filenames in your code. To distinguish between the two, be consistent and name variables that hold names of files "filename", "filepath" or "path", so you can use "file" consistently to denote a handle to a file.

Similarly, use "dirname", "dirpath", or "path" for variables holding directory names, to not get confused with a DIR * handle.

Move launch() and related functions into worker.c

Some worker-related functionality is in main.c instead of in worker.c. It's best to keep all the worker-related functionality together, and expose as little as possible of its internal workings. In main() I see:

worker_t worker;
worker_init(&worker, location);
int ret = launch(&worker, requeue_sec);
worker_destroy(&worker);

Unless there is a reason to split running the worker into three functions, I think it would be simpler if you could just write:

int ret = worker_launch(location, requeue_sec);

The function worker_launch() should live in worker.c, and basically do the same as worker_init(), launch() and worker_destroy().

Then in principle only worker_launch() has to be declared in worker.h, and the rest of the functions could be made static.

I see a similar pattern with tasks inside worker_run_once(). Perhaps you can do the same and just have task_run() call task_init() and task_destroy() internally?

Consider having the worker chdir() to the worker root

You are keeping track of a lot of directory names, and have to do a lot of path joining. It might make sense to just chdir(root_dir) in worker_init(). Then you don't have to store any of the directory names in worker_t.

Of course, you have to be careful that this does not violate assumptions in other parts of your program. However, since starting the worker is the only thing you do in main(), this seems fine.

Why does worker_requeue_check() need to run as a separate process?

I don't see why you need to run a separate process to requeue tasks. You can do this deterministically. Run worker_requeue_check() after every call to worker_run_once() inside the loop in worker_run_all(). If you want to avoid rescanning the directory so often, you can just check the time and avoid calling it if the last time you called it is less than requeue_interval / 2. You also have to call it once with a requeue_interval of 0 after the loop, to ensure that if the task directory is empty and worker_run_all() would exit, that it immediately requeues the failed tasks.

Alternatively, instead of rescanning requeue_dir every so often, you can keep a linked list of failed tasks along with the timestamp of failing. Then you just have to compare the timestamp of the first entry in the list to the current time to see if there is anything to requeue.

Don't use variable or function names starting with underscores

Identifiers starting with an underscore are reserved for use by the standard library. Not all combinations are reserved, but it's best to just not declare anything yourself that starts with one. So instead of __run_app(), just write run_app().

If you want to hide a symbol, the simplest way is to make it static. If it needs to be visible to other object files, but if you want to somehow indicate that some symbol is private, use underscores as a suffix.

Consider using posix_spawn() instead of fork()+execl()

The function posix_spawn() is made to start a new executable, and it can do a lot of housekeeping necessary to ensure the new process starts with a fresh environment. There are helper functions, like posix_spawn_file_actions_adddup2(), that allow you to redirect I/O.

Consider using asprintf() to build strings

You are already using a lot of Linux-specific functions, so consider using asprintf() to build a new string instead of doing a lot of low-level string manipulation. For example, filepath_join() can be rewritten as:

#define _GNU_SOURCE
#include <stdio.h>
...
char *filepath_join(const char *root_dir, const char *child) {
  char *data = NULL;
  asprintf(&data, "%s/%s", root_dir, child);
  return data;
}

Having two consecutive slashes in a filename is not a problem, so I wouldn't waste code on trying to remove an extraneous slash after root_dir.

Use basename() instead of writing your own function

Again, since you are already writing Linux-specific code, consider using basename() directly.

Make the path to the executable explicit

It's very weird to see code that gets the first executable in a directory. There is no guarantee in which order directories are traversed. So if there is more than one, it is not deterministic what your task runner is going to do. Why not pass the full filename to the executable to the worker?

Confusing use of brackets in usage()

In usage(), I see the following:

fprintf(stderr, "retask [flags]\n");
fprintf(stderr, "    -c <directory> [default=.]  use specified directory as work dir\n");
fprintf(stderr, "    -r <seconds>   [default=%li]  interval for requeue\n", DEFAULT_REQUEUE_INTERVAL);

In the first line, brackets are used to indicate optional arguments. That's indeed the standard way to represent those. But subsequent lines, you also write brackets around the default value. This is a bit confusing. I recommend you rewrite it to:

fprintf(stderr, "    -c <directory> Use specified directory as work dir. Default: %s\n", DEFAULT_LOCATION);
fprintf(stderr, "    -r <seconds>   Interval for requeue. Default: %i\n", DEFAULT_REQUEUE_INTERVAL);

Answer 3

I only want to add some nitpicks, but otherwise a really clean code base!

Don't create types with_t suffix

These types are reserved by POSIX and C is currently even moving to claiming some of these, too.

Use function prototypes

Currently you don't use function prototypes for functions with no arguments, eg.

void f();

This declaration defines the existence of a function returning void but "doesn't specify* whether it takes any arguments. To provide the necessary parameters (ie. that it does take no arguments), use:

void f(void);

This will allow the compiler to error-check for mismatching definitions/declarations.

Error handling: Print path

While it's a common pattern to eg. do

if (err = foo(), err == -1) {
    perror("progname: foo");
}
// Prints
// progname: foo: Some error message.

It's sensible to use a different pattern for functions like open:

if (fd = open("bla.txt", mode), fd == -1) {
    perror("progname: bla.txt");
}
// Prints eg.
// progname: bla.txt: File not found

You chose a kinda in-between approach, describing the specific file with "worker dir" or something similar, however this makes it hard to understand errors, eg. when the path to worker dir is ill-formed for some reason. I'd print that rather significant information for debugging as part of the message. The message about "worker dir creation failed" is more of a user-centered message. I'd split those (developer-centric, end-user-centric) to have this:

fprintf(stderr, "retask: %s: %s\n
        "Worker for creation failed\n", path, strerror(errno));

Declare variables when used

Since C99 you can declare variables later, when needed. This gives you far more possibilities for using const and often makes it easier to read as you don't need to "remember" what the variable was meant to say.

Vertical space

I'd use some empty lines to structure more, like paragraphs when writing prose. It makes it easier on the eyes.

Answer 4

You should consider to check the return value of malloc, on main.c you have

char *process_name = (char *) malloc(name_len + 1 + sizeof(REQUEUE_PROCESS));
memcpy(process_name, worker->name, name_len);

Consider that malloc could fail and you will have a problem.

char *process_name = (char *) malloc(name_len + 1 + sizeof(REQUEUE_PROCESS));
if (process_name != NULL) {
    memcpy(process_name, worker->name, name_len);