Concurrent Memory Pool System example: compile with `cc -std=gnu99 -I mps-1.114.0 mpsfmtgc.c mpsgc.c -lpthread`

mpsgc.c

#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <assert.h>
#include <unistd.h>
#include <pthread.h>

#include "mps.h"
#include "mpsavm.h"
#include "mpscamc.h"
#include "mpsgcfmt.h"

typedef struct {
  mps_arena_t arena;
  mps_fmt_t   obj_fmt;
  mps_chain_t obj_chain;
  mps_pool_t  obj_pool;
} mpsgc0_t;

typedef struct {
  mps_ap_t    obj_ap;
  mps_thr_t   thread;
  mps_root_t  reg_root;
} mpsgc0_thr_t;

static void
errorx(const char* str)
{
  fprintf(stderr, "%s\n", str);
  exit(1);
}

bool
mpsgc0_init(
  mpsgc0_t* gc,
  size_t arenasize,
  mps_gen_param_s* gen_params, size_t gen_params_len
)
{
  mps_res_t res;

  /* -- Create main MPS arena -- */
  MPS_ARGS_BEGIN(args) {
    MPS_ARGS_ADD(args, MPS_KEY_ARENA_SIZE, arenasize);
    res = mps_arena_create_k(&gc->arena, mps_arena_class_vm(), args);
  } MPS_ARGS_END(args);
  if (res != MPS_RES_OK) errorx("Couldn't create arena!");

  /* -- Create the object format descriptors -- */
  MPS_ARGS_BEGIN(args) {
    MPS_ARGS_ADD(args, MPS_KEY_FMT_ALIGN, ALIGNMENT);
    MPS_ARGS_ADD(args, MPS_KEY_FMT_SCAN, mpsgc0_obj_scan);
    MPS_ARGS_ADD(args, MPS_KEY_FMT_SKIP, mpsgc0_obj_skip);
    MPS_ARGS_ADD(args, MPS_KEY_FMT_ISFWD, mpsgc0_obj_isfwd);
    MPS_ARGS_ADD(args, MPS_KEY_FMT_FWD, mpsgc0_obj_fwd);
    MPS_ARGS_ADD(args, MPS_KEY_FMT_PAD, mpsgc0_obj_pad);
    res = mps_fmt_create_k(&gc->obj_fmt, gc->arena, args);
  } MPS_ARGS_END(args);
  if (res != MPS_RES_OK) errorx("Couldn't create obj format!");

  /* -- Create a chain controlling GC strategy -- */
  res = mps_chain_create(
    &gc->obj_chain, gc->arena, gen_params_len, gen_params);
  if (res != MPS_RES_OK) errorx("Couldn't create obj chain!");

  /* -- Create an AMC pool -- */
  MPS_ARGS_BEGIN(args) {
    MPS_ARGS_ADD(args, MPS_KEY_CHAIN, gc->obj_chain);
    MPS_ARGS_ADD(args, MPS_KEY_FORMAT, gc->obj_fmt);
    res = mps_pool_create_k(&gc->obj_pool, gc->arena, mps_class_amc(), args);
  } MPS_ARGS_END(args);
  if (res != MPS_RES_OK) errorx("Couldn't create obj pool!");

  /* Make sure we can pick up messages */
  //mps_message_type_enable(gc->arena, mps_message_type_finalization());
  //mps_message_type_enable(gc->arena, mps_message_type_gc());
  //mps_message_type_enable(gc->arena, mps_message_type_gc_start());

  return true;
}

void
mpsgc0_thr_init(mpsgc0_t* gc, mpsgc0_thr_t* thr, void* marker)
{
  mps_res_t res;

  /* -- Create an allocation point -- */
  res = mps_ap_create_k(&thr->obj_ap, gc->obj_pool, mps_args_none);
  if (res != MPS_RES_OK) errorx("Couldn't create allocation point!");

  /* -- Register current thread -- */
  res = mps_thread_reg(&thr->thread, gc->arena);
  if (res != MPS_RES_OK) errorx("Couldn't register thread");

  /* -- Create root -- */
  res = mps_root_create_reg(
    &thr->reg_root, gc->arena, mps_rank_ambig(), 0, thr->thread,
    mps_stack_scan_ambig, marker, 0);
  if (res != MPS_RES_OK) errorx("Couldn't create root");
}

void
mpsgc0_thr_fini(mpsgc0_t* gc, mpsgc0_thr_t* thr)
{
  /* -- Clean up thread-specific objects -- */
  mps_arena_park(gc->arena);
  mps_root_destroy(thr->reg_root);
  mps_thread_dereg(thr->thread);
  mps_ap_destroy(thr->obj_ap);
  mps_arena_release(gc->arena);
}

void
mpsgc0_fini(mpsgc0_t* gc)
{
  /* -- Clean up all MPS objects -- */
  mps_arena_park(gc->arena);
  mps_pool_destroy(gc->obj_pool);
  mps_chain_destroy(gc->obj_chain);
  mps_fmt_destroy(gc->obj_fmt);
  mps_arena_destroy(gc->arena);
}

static mpsgc0_obj_t
mpsgc0_make_integer(mps_ap_t obj_ap, long integer)
{
  mpsgc0_obj_t obj;
  mps_addr_t addr;
  size_t size = ALIGN_OBJ(sizeof(mpsgc0_integer_s));

  do {
    mps_res_t res = mps_reserve(&addr, obj_ap, size);
    if (res != MPS_RES_OK) errorx("out of memory in make_integer");
    obj = addr;
    obj->integer.type = TYPE_INTEGER;
    obj->integer.integer = integer;
  } while(!mps_commit(obj_ap, addr, size));
  return obj;
}

static mpsgc0_obj_t
mpsgc0_make_string(mps_ap_t obj_ap, size_t length, const char *string)
{
  mpsgc0_obj_t obj;
  mps_addr_t addr;
  size_t size = ALIGN_OBJ(offsetof(mpsgc0_string_s, string) + length + 1);

  do {
    mps_res_t res = mps_reserve(&addr, obj_ap, size);
    if (res != MPS_RES_OK) errorx("out of memory in make_string");
    obj = addr;
    obj->string.type = TYPE_STRING;
    obj->string.len  = length;

    if (string) memcpy(obj->string.string, string, length+1);
    else memset(obj->string.string, 0, length+1);

  } while(!mps_commit(obj_ap, addr, size));
  return obj;
}

/* -------------------------------------------------------------------------- */
/* -- Chat utility for MPS messages ----------------------------------------- */

static void
mpsgc0_chat(mpsgc0_t* gc)
{
  mps_message_type_t type;

  while (mps_message_queue_type(&type, gc->arena)) {
    mps_message_t message;
    mps_bool_t b;
    b = mps_message_get(&message, gc->arena, type);
    assert(b); /* we just checked there was one */

    if (type == mps_message_type_gc_start()) {
      printf("Collection started.\n");
      printf("  Why: %s\n", mps_message_gc_start_why(gc->arena, message));
      printf("  Clock: %lu\n", (unsigned long)mps_message_clock(gc->arena, message));

    } else if (type == mps_message_type_gc()) {
      size_t live = mps_message_gc_live_size(gc->arena, message);
      size_t condemned = mps_message_gc_condemned_size(gc->arena, message);
      size_t not_condemned = mps_message_gc_not_condemned_size(gc->arena, message);
      printf("Collection finished.\n");
      printf("    live %lu\n", (unsigned long)live);
      printf("    condemned %lu\n", (unsigned long)condemned);
      printf("    not_condemned %lu\n", (unsigned long)not_condemned);
      printf("    clock: %lu\n", (unsigned long)mps_message_clock(gc->arena, message));

    } else {
      printf("Unknown message from MPS!\n");
    }

    mps_message_discard(gc->arena, message);
  }
}

/* -------------------------------------------------------------------------- */
/* -- Main driver program --------------------------------------------------- */

static int __attribute__ ((noinline))
__mainentry_start(mpsgc0_t* gc, mpsgc0_thr_t* thr, int argc, char** av)
{
  mpsgc0_obj_t obj;

  for(;;) {
    usleep(1*1000*100); // 0.1 seconds
    mpsgc0_chat(gc);
#define STR "hello world omfg"
    for (int i = 0; i < 1024; i++) {
      obj = mpsgc0_make_string(thr->obj_ap, strlen(STR), STR);
    }
#undef STR
    printf("%lu: Hello world, allocated 1024 string objects\n", pthread_self());
    fflush(stdout);
  }

  return 0;
}

static void*
threadstart(void* gcptr)
{
  void* marker = &marker;
  mpsgc0_t* gc = (mpsgc0_t*)gcptr;
  mpsgc0_thr_t thr;

  mpsgc0_thr_init(gc, &thr, marker);
  __mainentry_start(gc, &thr, 0, NULL);
  mpsgc0_thr_fini(gc, &thr);

  return NULL;
}

/**
 * main() test program
 */
int
main(int ac, char** av)
{
  mpsgc0_t gc;
  size_t arenasize = 32ul * 1024 * 1024;
  mps_gen_param_s mpsgc0_gen_params[] = {
    { 150, 0.85 },
    { 170, 0.45 }
  };

  pthread_t* threads = NULL;
  threads = malloc(5*sizeof(pthread_t));

  /* Start GC */
  mpsgc0_init(&gc, arenasize, mpsgc0_gen_params, 2);

  /* Create threads */
  for (int i = 0; i < 5; i++) {
    if (pthread_create(&threads[i], NULL, threadstart, &gc))
      errorx("Couldn't create thread!");
  }

  /* Wait for threads to finish */
  for (int i = 0; i < 5; i++) {
    pthread_join(threads[i], NULL);
  }

  /* Done */
  free(threads);
  mpsgc0_fini(&gc);
  return 0;
}

mpsgcfmt.c

#include "mps.c"
#include "mpsgcfmt.h"

/**
 * Scan method
 */
mps_res_t
mpsgc0_obj_scan(mps_ss_t ss, mps_addr_t base, mps_addr_t limit)
{
#define FIX(ref)						    \
  do {								    \
    mps_addr_t _addr = (ref); /* copy to local to avoid type pun */ \
    mps_res_t _res = MPS_FIX12(ss, &_addr);			    \
    if (_res != MPS_RES_OK) return _res;			    \
    (ref) = _addr;						    \
  } while(0)

  MPS_SCAN_BEGIN(ss) {
    while (base < limit) {
      mpsgc0_obj_t obj = base;

      switch (MPSGC0_TYPE(obj)) {

      case TYPE_INTEGER:

      case TYPE_STRING:
	base = (char *)base +
          ALIGN_OBJ(offsetof(mpsgc0_string_s, string) + obj->string.len + 1);
        break;
      case TYPE_FWD2:
        base = (char *)base + ALIGN_WORD(sizeof(mpsgc0_fwd2_s));
        break;
      case TYPE_FWD:
        base = (char *)base + ALIGN_WORD(obj->fwd.size);
        break;
      case TYPE_PAD1:
        base = (char *)base + ALIGN_WORD(sizeof(mpsgc0_pad1_s));
        break;
      case TYPE_PAD:
        base = (char *)base + ALIGN_WORD(obj->pad.size);
        break;
      default:
	assert("Invalid object type marked for scanning!" && 0);
	abort();
      }
    }
  } MPS_SCAN_END(ss);

  return MPS_RES_OK;
}

mps_addr_t
mpsgc0_obj_skip(mps_addr_t base)
{
  mpsgc0_obj_t obj = base;

  switch(MPSGC0_TYPE(obj)) {
  case TYPE_INTEGER:
    base = (char *)base + ALIGN_OBJ(sizeof(mpsgc0_integer_s));
    break;
  case TYPE_STRING:
    base = (char *)base +
      ALIGN_OBJ(offsetof(mpsgc0_string_s, string) + obj->string.len + 1);
    break;
  case TYPE_FWD2:
    base = (char *)base + ALIGN_WORD(sizeof(mpsgc0_fwd2_s));
    break;
  case TYPE_FWD:
    base = (char *)base + ALIGN_WORD(obj->fwd.size);
    break;
  case TYPE_PAD:
    base = (char *)base + ALIGN_WORD(obj->pad.size);
    break;
  case TYPE_PAD1:
    base = (char *)base + ALIGN_WORD(sizeof(mpsgc0_pad1_s));
    break;
  default:
    assert("Unexpected object on the heap!" && 0);
    abort();
  }

  return base;
}

mps_addr_t
mpsgc0_obj_isfwd(mps_addr_t addr)
{
  mpsgc0_obj_t obj = addr;

  switch (MPSGC0_TYPE(obj)) {
  case TYPE_FWD2:
    return obj->fwd2.fwd;
  case TYPE_FWD:
    return obj->fwd.fwd;
  default:
    assert("Unexpected object checked as forwarding pointer!" && 0);
    abort();
  }

  return NULL;
}

void
mpsgc0_obj_fwd(mps_addr_t old, mps_addr_t new)
{
  mpsgc0_obj_t obj = old;
  mps_addr_t limit = mpsgc0_obj_skip(old);

  size_t size = (size_t)((char *)limit - (char *)old);
  assert(size >= ALIGN_WORD(sizeof(mpsgc0_fwd2_s)));

  if (size == ALIGN_WORD(sizeof(mpsgc0_fwd2_s))) {
    MPSGC0_TYPE(obj) = TYPE_FWD2;
    obj->fwd2.fwd = new;
  } else {
    MPSGC0_TYPE(obj) = TYPE_FWD;
    obj->fwd.fwd  = new;
    obj->fwd.size = size;
  }
}

void
mpsgc0_obj_pad(mps_addr_t addr, size_t size)
{
  mpsgc0_obj_t obj = addr;

  assert(size >= ALIGN_WORD(sizeof(mpsgc0_pad1_s)));
  if (size == ALIGN_WORD(sizeof(mpsgc0_pad1_s))) {
    MPSGC0_TYPE(obj) = TYPE_PAD1;
  } else {
    MPSGC0_TYPE(obj) = TYPE_PAD;
    obj->pad.size = size;
  }
}