Fast File Search

We’ve covered the major parts of the pthread library, but it will still take a lot of practice to implement them not only correctly, but in a way that speeds up a program.

I find learning multi-threading to be particularly challenging because most programs can’t easily benefit from more threads. Almost every problem can be solved with more threads, but that doesn’t mean that it will make it any faster. Threads are not free, and they bring along a lot of overhead along with them. This is why it is important that you benchmark your code with and without threads to decide which variant is faster. You may be surprised by how much a single thread of execution can do on its own.

Of course, our 100000 byte statically allocated buffer will probably work just fine for searching most files, but it won’t cover every possible case. We’ll create a dynamically allocated buffer to take care of this.

#define MAX(a, b) ((a) > (b) ? (a) : (b))

typedef struct buffer {
  char *buffer;
  size_t len;
  size_t cap;
} buffer;

void buffer_resize(buffer *b, size_t added_size) {
  if (b->len + added_size > b->cap) {
    b->cap += MAX(4096, added_size);
    b->buffer = realloc(b->buffer, b->cap);
    if (!b->buffer) {
      perror("realloc");
      exit(1);
    }
  }
}

void buffer_write(buffer *b, const char *format, ...) {
  va_list args;
  va_start(args, format);
  int length = vsnprintf(NULL, 0, format, args) + 1;
  va_end(args);
  va_start(args, format);
  buffer_resize(b, length );
  b->len += vsprintf(&b->buffer[b->len], format, args);
  b->buffer[b->len] = '\0';
  va_end(args);
}

We can use this buffer to replace the static array from before.

void file_search(const char *const filename, const char *const needle) {
  buffer b = (buffer){.buffer = malloc(128), .len = 0, .cap = 128};

  buffer_write(&b, "%s%s%s:\n", CYAN_BOLD, filename, RESET);
// ....
    char *prev = buf;
    char *found = strcasestr(buf, needle);
    if (found) {
      buffer_write(&b, "%s%lu%s: ", CYAN_BOLD, line_number, RESET);
    }
    while (found) {
// ....
      size_t len_before = found - prev;
      buffer_write(&b, "%.*s", (int)len_before, prev);
      buffer_write(&b, "%s%.*s%s", RED_BOLD, (int)strlen(needle), found, RESET);
      prev = found + strlen(needle);
// ....
      if (!found) {
        buffer_write(&b, "%s", prev);
      }
// ....
  if (found_something) {
    pthread_mutex_lock(&io_mutex);
    printf("%s\n", b.buffer);
    pthread_mutex_unlock(&io_mutex);
  }
// ....
  free(b.buffer);
  free(buf);
  return;
}

Our first step is to track all the work that needs to be done. Each thread will take different amounts of time to parse their respective files, so we place all the work on the queue, and then lets the threads pull work off as they are free. This is a very specific implementation of a queue, and its behavior specifically applies to this program. A more generic queue can be used if desired.

struct node {
  struct node *next;
  char *name;
};
typedef struct node node_t;

node_t *head = NULL;
node_t *tail = NULL;

void enqueue(char *entry) {
  node_t *newnode = malloc(sizeof(node_t));
  newnode->name = entry;
  newnode->next = NULL;
  if (tail == NULL) {
    head = newnode;
  } else {
    tail->next = newnode;
  }
  tail = newnode;
}

// dequeue returns dirent* or NULL
char *dequeue(void) {
  if (head == NULL) {
    return NULL;
  } else {
    char *result = head->name;
    node_t *temp = head;
    head = head->next;
    if (head == NULL) {
      tail = NULL;
    }
    free(temp);
    return result;
  }
}

The basic premise here is that a queue has a front and a back, called head and tail the above code. Items are placed at the tail of the list, and removed from the front of the list, that is the entire interface. When an item is inserted we allocate memory for its node, and place it into the list by updating the tail pointer, or the header pointer if the list is empty.

When an item is removed from the list we want to return the string value directly, but we first grab a handle to the node, so we can free its memory. From there we update the links in the list and return the value. In the case the list is empty we return NULL, which signifies that our work queue is empty.

Our program will require 2 pieces of synchronization between our threads. We need to have a mutex for our work queue, and a mutex for printing to the terminal. The queue mutex ensures that only one job is pulled off the queue at a time, and that jobs are added to the queue atomically. The input/output mutex ensures that an entire threads output is printed cleanly.

Luckily, our program is a special case where all the work is placed into the queue up front, and then our thread pool just needs to go through the work until it is empty. Therefore, we won’t have to use condition variables to alert a thread there is work to be done, we can just sit and spin until work shows up. Once again, this only works in short-lived programs where we do not anticipate threads sitting around and waiting for work (as in a web server).

pthread_mutex_t q_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t io_mutex = PTHREAD_MUTEX_INITIALIZER;
bool terminate_work = false;

We create 2 mutexes, and a third global variable to track when we should terminate work. This variable will be used by our thread_function telling it to spin while terminate_work is false, and terminate once it is true.

In our case the “work” that we want to do is scan files, so each item in the queue should be a path to a file. Like we said before, we want to scan the current directory for files, and recursively scan its subdirectories. In C we get the contents of a directory through opendir, which through the use of readdir gives a pointer to each item in the directory. You can think of this as running ls in the terminal, and being given back a pointer to a struct representing each of the items.

// Recursively scans a directory placing all files
// on the work queue
void handle_dir(const char *dir_path) {
  DIR *dir;
  struct dirent *entry;

  dir = opendir(dir_path);
  if (!dir) {
    perror("opendir");
    exit(1);
  }
// continued below
// ....

We start with a directory path as a string, which will be "." to start, indicating the current directory where the program is being run. The directory is then opened and error checked. Then we need to walk through each entry in the directory, placing it directly on the work queue if it is a file, or recursively scanning if it is a directory.

// ....
// continued from above
  while ((entry = readdir(dir))) {
    char path[PATH_MAX];
    sprintf(path, "%s/%s", dir_path, entry->d_name);
    if (entry->d_type == DT_REG) {
      pthread_mutex_lock(&q_mutex);
      enqueue(strdup(path));
      pthread_mutex_unlock(&q_mutex);
    } else if (entry->d_type == DT_DIR && strcmp(entry->d_name, ".") != 0 && strcmp(entry->d_name, "..") != 0) {
      handle_dir(path);
    }
  }

  closedir(dir);
}

Calling readdir(dir) yields the next entry from the directory. We assign this to the entry variable, and continue looping while its value is non-null. Our next step is to create a string representation of its path, which we can do by appending the name of the entry onto the current path. This means the file temp.txt in a subdirectory foo will result in a path of ./foo/temp.txt.

If we are looking at a file we push its path onto the work queue by creating a copy of the path. Otherwise, if we’re looking at a subdirectory we call ourselves again to repeat the process until we get to a directory with nothing but files.

You may notice that we acquire the q_mutex, then add our work on the queue, before finally releasing the lock. We will see why this is necessary later in main. For now, it is important to know that even though a queue has 2 ends which seem independent, when a queue has 1 element or is empty the ends begin to interact with each other. So when calling either enqueue or dequeue we need to lock the q_mutex.

Other than adding an automatically resizing buffer we also need to make one small change to this function, simply locking the io_mutex before printing, and releasing it once we’re done.

// ....
  if (found_something) {
    pthread_mutex_lock(&io_mutex);
    printf("%.*s\n", (int)output_len, output);
    pthread_mutex_unlock(&io_mutex);
  }
// ....

Finally we can get to doing the work in our threads. Upon being created, each thread will enter a loop where they check for work, loop if none is found, and process a file if one is available.

We will also use our terminate_work flag here to signal to a thread that if there is no work on the queue, and you are being told to terminate work, then you can go ahead and exit.

void *thread_function(void *arg) {
  char *needle = (char*)arg;
  while (true) {
    char *filename;
    pthread_mutex_lock(&q_mutex);
    if (!terminate_work && head == NULL) {
      pthread_mutex_unlock(&q_mutex);
      continue;
    }
    if (terminate_work && head == NULL) {
      pthread_mutex_unlock(&q_mutex);
      break;
    }

    filename = dequeue();

    pthread_mutex_unlock(&q_mutex);
    if (filename) {
      file_search(filename, needle);
      free(filename);
    }
  }
  pthread_exit(NULL);
}

Each thread will acquire the q_mutex and check for work by seeing if head is NULL. Once a thread has the lock it is free to check and/or modify any of the shared structures that mutex is protecting. In our case we make a decision based upon the value of terminate_work and the presence of work in the queue.

If a thread finds work it can grab that filename off of the queue, and then release the lock allowing other threads to continue execution. By first acquiring the lock we guarantee that the state of the queue will not change until we release it. This is why it is so important that both enqueue and dequeue are only called while the q_mutex is held.

Our 2 conditional checks have different behavior. If we want our thread to continue working we call continue, otherwise if we want to exit we break out of the infinite loop. This lands our thread at pthread_exit(NULL), which allows the thread to now be joined on from main. Telling our threads to continue can be wasteful, but we will only be in this state for a very short period of time before work is placed onto the queue.

The last thing to note here is that since each filename was cloned with strdup, its memory needs to be freed. After we call file_search we no longer need that filename, and we’re safe to free it.

Now we can work to put everything together in main.

We can start out with a simple usage statement that helps our users use the CLI.

int main(int argc, char **argv) {
  if (argc < 2) {
    printf("usage: %s needle", argv[0]);
    exit(1);
  }
// continued below
// ....

The string we are searching for is called our “needle”, and it is expected to the in argv[1]. We want to pass this needle to each thread, so we store it in a variable to give it some semantic meaning aside from argv[1].

// ....
  char *needle = argv[1];
// ....

As I said before, it is common to scale the number of threads to the number of CPUs available on a machine. When our program starts up it will ask the operating system how many CPUs is has, and then we create a thread pool of that size.

// ....
  long num_cpu = sysconf(_SC_NPROCESSORS_CONF);
  if (num_cpu <= 0) {
    fprintf(stderr, "Error determining the number of CPUs.\n");
    exit(1);
  }

  pthread_t *thread_pool = alloca(num_cpu * sizeof(pthread_t));
  for (int i = 0; i < num_cpu; i++) {
    pthread_create(&thread_pool[i], NULL, thread_function, (void*)needle);
  }
// ....

After calling sysconf we have the number of CPUs available, which we error check to ensure it is a value that makes sense.

We then create a thread pool of type pthread_t*, which has its memory allocated with alloca. alloca is another allocation function similar to malloc, but instead of getting memory from the heap, we dynamically get memory from the stack. It is a very interesting function that allows for a stack allocation of a size that is only known at runtime. Since the memory resides on the stack, that also means that there is no need to call a free function, the memory is automatically cleaned up.

Each thread is created with our thread function, with their only argument being the needle itself. To avoid any type warnings we must cast our needle to a void* since that is what the type the thread_function accepts. We will type cast this back to a char* in our thread function.

Now all of our threads are spinning, waiting for work, so let’s give them something to work on.

// ....
  handle_dir(".");
  terminate_work = true;
// ....

Because handle_dir is recursive, we only need to give it an entry point, and it does all the work from there. This leaves our main function very clean, only having to specify the current directory with ".".

The work queue is now being filled up, and at the same time the threads are churning through the files as fast as they can. As soon as handle_dir returns we know all the work has been added to the queue, at that point we can tell our threads to terminate work. However, they won’t immediately stop working. Our thread_function will only exit when it is told to terminate and there is no work left on the queue.

If we finished main off now we would be missing some of our output, so we need to wait for each thread to finish working. We can do that with pthread_join.

// ....
  for (int i = 0; i < num_cpu; i++) {
    pthread_join(thread_pool[i], NULL);
  }

}
// end of main

That’s it! No need to clean up any memory from our thread pool because it was stack allocated!

Let’s see how this works in a real directory structure.

We can compare that to ripgrep, which is a program that has a mode with very similar features to our implementation.

Speaking of ripgrep it has another feature that allows it to filter any input given to it on stdin

[jarch@jarch jacksonmowry.github.io]$ ls | rg ".txt"
analysis.txt
needle.txt

We should be able to adapt our program to match this functionality.

Apps that render any sort of live user interface often benefit from separating rendering and application logic to different threads. This way any heavy computation on one thread does not slow the user interface to a crawl. Chat apps may have one thread to listen for incoming messages, another to send messages, and yet another to render the messages.

This is the category that our FFS program falls under. Each input to our find function is an entirely separate file, they share nothing in common other than their location in the file system. If you have used ripgrep before, or were following along and running the code here, you may have noticed that the output from either is non-deterministic. This is because threads are scheduled by the operating system, and when a thread is spawned it may be backgrounded for a brief time, while another more important thread is run.

Each time the program is run we may get a different ordering of files. We are ok with this tradeoff because each file it treated as its own entity, which is searched independently of other files. That is not to say that programs requiring a specific output order cannot also use multi-threading, those problems just require extra work to maintain ordering.

If a single-threaded program has numerous inputs that are currently being processed one-by-one with no relation to the previous item processed, it may be a good candidate for multithreaded.

If each item in a large set is being processed one after the other, there may be a chance to split that set up in to smaller sets for processing. This is a hard one to get right, as it requires much more careful control over mutual exclusion, which may slow the program down.

Our program also falls into this category, to a much lesser extent. In terms of speed reading from a file, or printing output to the terminal are much slower than the rest of the program. So a program may benefit from having separate threads of execution so that while one thread is waiting on a file to open, another can be doing a heavy computation.

Most servers that communicate over a socket architecture will benefit from having more handlers for those sockets. Imagine a web server able to handle 1 concurrent connection, versus a server able to handle 12 concurrent connections. Not only will these speed up average response times for you users, you will also allow your code to scale much better across different architectures.

Hopefully we will be building a simple socket server coming up very soon!