Also take flock locks

[tl-append] / tl-append-test.c

#define _POSIX_C_SOURCE 2
#define _XOPEN_SOURCE
#define _GNU_SOURCE
#include <assert.h>
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/file.h>
#include <time.h>
#include <unistd.h>

#include "common.h"

const size_t TIMESTAMP_LEN = 19;

typedef struct expectation {
  time_t a, b;
  const char *message;
} ex_t;

const ex_t END = {((time_t)-1), ((time_t)-1), NULL};
const ex_t CONSUMED = {((time_t)-2), ((time_t)-2), NULL};
static int is_end(ex_t exp) {
  return exp.a == END.a && exp.b == END.b && exp.message == END.message;
}
static int is_consumed(ex_t exp) {
  return exp.a == CONSUMED.a && exp.b == CONSUMED.b &&
         exp.message == CONSUMED.message;
}
static ex_t expectation(time_t a, time_t b, const char *message) {
  ex_t exp;
  exp.a = a;
  exp.b = b;
  exp.message = message;
  return exp;
}

static void remove_logfile() {
  if (remove(FILENAME) != 0) {
    if (errno != ENOENT) {
      die_err("Error removing log file");
    }
  }
}

static ex_t write_to_tl_append(const char *content) {
  FILE *p = popen("./tl-append", "w");
  if (p == NULL)
    die_err("Couldn't run tl-append");
  time_t start = time(NULL);
  if (fputs(content, p) == EOF)
    die("Couldn't write to pipe");
  int status = pclose(p);
  time_t end = time(NULL);
  if (status < 0)
    die_err("Error closing pipe");
  if (status != 0)
    die("tl-append exited abnormally");
  return expectation(start, end, content);
}

static char *timestamp_problem(const ex_t *ex, const char *line) {
  struct tm tm;

  /* localtime_r to set tm's timezone */
  time_t now_time = time(NULL);
  if (localtime_r(&now_time, &tm) == NULL)
    return "Can't unpack current time?";

  const char *strptime_result = strptime(line, "%Y %m %d %H %M %S", &tm);
  if (strptime_result == NULL || strptime_result != &line[TIMESTAMP_LEN])
    return "Wrong contents in log file: Couldn't parse timestamp";
  time_t t = mktime(&tm);
  int t_in_range = ex->a <= t && t <= ex->b;
  if (!t_in_range)
    return "Wrong contents in log file: Wrong time";
  return NULL;
}

static char *line_problem(const ex_t *ex, const char *line) {
  size_t line_len = strlen(line);
  if (line_len < TIMESTAMP_LEN + 1)
    return "Wrong contents in log file: Line too short";
  char *trouble = timestamp_problem(ex, line);
  if (trouble)
    return trouble;
  if (line[TIMESTAMP_LEN] != ' ')
    return "Wrong contents in log file: Bad format";
  if (strncmp(ex->message, &line[TIMESTAMP_LEN + 1], strlen(ex->message) + 1) !=
      0)
    return "Wrong contents in log file";
  return NULL;
}

static void verify_line(const ex_t *ex, const char *line) {
  char *trouble = line_problem(ex, line);
  if (trouble)
    die(trouble);
}

static void verify_log_contents(const ex_t exps[]) {
  FILE *f = fopen(FILENAME, "r");
  if (f == NULL)
    die_err("Error opening log file");
  for (size_t i = 0; !is_end(exps[i]); i++) {
    size_t len = TIMESTAMP_LEN + 1 + strlen(exps[i].message);
    if (len > INT_MAX - 1)
      die("message too long");
    char *buf = (char *)malloc(len + 2);
    if (fgets(buf, len + 1, f) == NULL)
      die("Error reading log file");
    if (ferror(f))
      die("Error reading log file");
    verify_line(&exps[i], buf);
    free(buf);
  }
  if (fclose(f) != 0)
    die_err("Error closing log file");
}

static void consume_expectation(ex_t exps[], const char *line) {
  for (size_t i = 0; !is_end(exps[i]); i++) {
    if (line_problem(&exps[i], line) == NULL) {
      exps[i] = CONSUMED;
      return;
    }
  }
  die("Wrong contents in log file: Line didn't match any expectation");
}

static void verify_log_contents_unordered(ex_t exps[]) {
  FILE *f = fopen(FILENAME, "r");
  if (f == NULL)
    die_err("Error opening log file");
  for (;;) {
    const int MAX_LEN = 9999;
    char buf[MAX_LEN];
    if (fgets(buf, MAX_LEN, f) == NULL) {
      if (feof(f))
        break;
      die("Error reading log file");
    }
    if (ferror(f))
      die("Error reading log file");
    consume_expectation(exps, buf);
  }
  if (fclose(f) != 0)
    die_err("Error closing log file");
  for (size_t i = 0; !is_end(exps[i]); i++) {
    if (!is_consumed(exps[i]))
      die("Wrong contents in log file: Unconsumed expectation");
  }
}

static void test_encode_time() {
  /* localtime_r to set tm's timezone */
  time_t now_time = time(NULL);
  struct tm tm;
  if (localtime_r(&now_time, &tm) == NULL)
    die_err("Can't unpack current time?");

  const char *strptime_result = strptime("2011-12-13 14:15:16", "%F %T", &tm);
  if (strptime_result == NULL || *strptime_result != '\0')
    die("Couldn't parse time?");
  time_t tt = mktime(&tm);
  if (tt == (time_t)-1)
    die_err("Can't pack time?");

  const char *encoded = encode_time(tt);
  /* Loose check to allow for daylight savings time changes between the current
   * time and the target time.  :( */
  assert(encoded[0] == '2');
  assert(encoded[1] == '0');
  assert(encoded[2] == '1');
  assert(encoded[3] == '1');
  assert(encoded[4] == ' ');
  assert(encoded[5] == '1');
  assert(encoded[6] == '2');
  assert(encoded[7] == ' ');
  assert(encoded[8] == '1');
  assert(encoded[9] == '3');
  assert(encoded[10] == ' ');
  assert(encoded[11] == '1');
  assert(isdigit(encoded[12]));
  assert(encoded[13] == ' ');
  assert(isdigit(encoded[14]));
  assert(isdigit(encoded[15]));
  assert(encoded[16] == ' ');
  assert(encoded[17] == '1');
  assert(encoded[18] == '6');
  assert(encoded[19] == '\0');
}

static FILE *take_lock(FILE *f, char *lock_type) {
  int fd = fileno(f);
  if (fd == -1)
    die_err("Couldn't get file descriptor for locking");
  if (strcmp(lock_type, "fcntl") == 0) {
    if (fcntl(fd, F_SETLK,
              &(struct flock){.l_type = F_WRLCK,
                              .l_whence = SEEK_SET,
                              .l_start = 0,
                              .l_len = 0}) == -1)
      die_err("Couldn't take fcntl lock");
  } else if (strcmp(lock_type, "flock") == 0) {
    if (flock(fd, LOCK_EX) == -1)
      die_err("Couldn't take flock lock");
  } else {
    die("Bad lock type");
  }
  return f;
}

static void release_lock(FILE *f) {
  if (fclose(f) != 0)
    die_err("Error releasing lock");
}

static void *release_lock_after_delay(void *f) {
  sleep(1);
  release_lock((FILE *)f);
  return NULL;
}

static void *writer_thread(void *start_signal) {
  ex_t *ex = (ex_t *)malloc(sizeof(ex_t));
  if (ex == NULL)
    die_err("Couldn't allocate memory");
  char *message;
  if (asprintf(&message, "Hello from thread %lu\n", pthread_self()) == -1)
    die("Couldn't prepare message");
  pthread_rwlock_rdlock((pthread_rwlock_t *)start_signal);
  *ex = write_to_tl_append(message);
  return ex;
}

static void write_and_read_line() {
  remove_logfile();
  ex_t e = write_to_tl_append("foo\n");
  verify_log_contents((ex_t[]){e, END});
}

static void write_and_read_two_lines() {
  remove_logfile();
  ex_t e1 = write_to_tl_append("foo\n");
  ex_t e2 = write_to_tl_append("bar\n");
  verify_log_contents((ex_t[]){e1, e2, END});
}

static void write_to_locked_log(char *lock_types[]) {
  remove_logfile();
  ex_t e1 = write_to_tl_append("begin\n");
  FILE *f = fopen(FILENAME, "ae");
  if (f == NULL)
    die_err("Couldn't open file for locking");
  for (int i = 0; lock_types[i]; i++)
    take_lock(f, lock_types[0]);
  pthread_t unlock_thread;
  int create_ret =
      pthread_create(&unlock_thread, NULL, &release_lock_after_delay, f);
  if (create_ret != 0) {
    errno = create_ret;
    die_err("Couldn't start thread");
  }
  ex_t e2 = write_to_tl_append("delayed\n");
  int join_ret = pthread_join(unlock_thread, NULL);
  if (join_ret != 0) {
    errno = join_ret;
    die_err("Couldn't join thread");
  }
  verify_log_contents((ex_t[]){e1, e2, END});
}

static void write_concurrently() {
  remove_logfile();
  const int PARALLELISM = 250;
  pthread_t threads[PARALLELISM];
  pthread_rwlock_t start_signal;
  pthread_rwlock_init(&start_signal, NULL);
  for (int i = 0; i < PARALLELISM; i++) {
    int create_ret =
        pthread_create(&threads[i], NULL, &writer_thread, &start_signal);
    if (create_ret != 0) {
      errno = create_ret;
      die_err("Couldn't start thread");
    }
  }
  pthread_rwlock_unlock(&start_signal);
  ex_t results[PARALLELISM + 1];
  for (int i = 0; i < PARALLELISM; i++) {
    ex_t *ex;
    int join_ret = pthread_join(threads[i], (void **)&ex);
    if (join_ret != 0) {
      errno = join_ret;
      die_err("Couldn't join thread");
    }
    results[i] = *ex;
    free(ex);
  }
  results[PARALLELISM] = END;
  verify_log_contents_unordered(results);
}

int main() {
  test_encode_time();
  write_and_read_line();
  write_and_read_two_lines();
  write_to_locked_log((char *[]){NULL});
  write_to_locked_log((char *[]){"fcntl", NULL});
  write_to_locked_log((char *[]){"flock", NULL});
  write_to_locked_log((char *[]){"flock", "fcntl", NULL}); /* Deadlock risk! */
  write_to_locked_log((char *[]){"fcntl", "flock", NULL});
  write_concurrently();
}