c_cpp/lib/klib/test/kthread_test2.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <string.h>
   4 #include <assert.h>
   5
   6 void kt_for(int n_threads, void (*func)(void*,long,int), void *data, long n);
   7 void kt_pipeline(int n_threads, void *(*func)(void*, int, void*), void *shared_data, int n_steps);
   8
   9 typedef struct {
  10         FILE *fp;
  11         int max_lines, buf_size, n_threads;
  12         char *buf;
  13 } pipeline_t;
  14
  15 typedef struct {
  16         int n_lines;
  17         char **lines;
  18 } step_t;
  19
  20 static void worker_for(void *_data, long i, int tid) // kt_for() callback
  21 {
  22         step_t *step = (step_t*)_data;
  23         char *s = step->lines[i];
  24         int t, l, j;
  25         l = strlen(s) - 1;
  26         assert(s[l] == '\n'); // not supporting long lines
  27         for (j = 0; j < l>>1; ++j)
  28                 t = s[j], s[j] = s[l - 1 - j], s[l - 1 - j] = t;
  29 }
  30
  31 static void *worker_pipeline(void *shared, int step, void *in) // kt_pipeline() callback
  32 {
  33         pipeline_t *p = (pipeline_t*)shared;
  34         if (step == 0) { // step 0: read lines into the buffer
  35                 step_t *s;
  36                 s = calloc(1, sizeof(step_t));
  37                 s->lines = calloc(p->max_lines, sizeof(char*));
  38                 while (fgets(p->buf, p->buf_size, p->fp) != 0) {
  39                         s->lines[s->n_lines] = strdup(p->buf);
  40                         if (++s->n_lines >= p->max_lines)
  41                                 break;
  42                 }
  43                 if (s->n_lines) return s;
  44         } else if (step == 1) { // step 1: reverse lines
  45                 kt_for(p->n_threads, worker_for, in, ((step_t*)in)->n_lines);
  46                 return in;
  47         } else if (step == 2) { // step 3: write the buffer to output
  48                 step_t *s = (step_t*)in;
  49                 while (s->n_lines > 0) {
  50                         fputs(s->lines[--s->n_lines], stdout);
  51                         free(s->lines[s->n_lines]);
  52                 }
  53                 free(s->lines); free(s);
  54         }
  55         return 0;
  56 }
  57
  58 int main(int argc, char *argv[])
  59 {
  60         pipeline_t pl;
  61         int pl_threads;
  62         if (argc == 1) {
  63                 fprintf(stderr, "Usage: reverse <in.txt> [pipeline_threads [for_threads]]\n");
  64                 return 1;
  65         }
  66         pl.fp = strcmp(argv[1], "-")? fopen(argv[1], "r") : stdin;
  67         if (pl.fp == 0) {
  68                 fprintf(stderr, "ERROR: failed to open the input file.\n");
  69                 return 1;
  70         }
  71         pl_threads = argc > 2? atoi(argv[2]) : 3;
  72         pl.max_lines = 4096;
  73         pl.buf_size = 0x10000;
  74         pl.n_threads = argc > 3? atoi(argv[3]) : 1;
  75         pl.buf = calloc(pl.buf_size, 1);
  76         kt_pipeline(pl_threads, worker_pipeline, &pl, 3);
  77         free(pl.buf);
  78         if (pl.fp != stdin) fclose(pl.fp);
  79         return 0;
  80 }