<commit_before>/*
 * Copyright 2013 University of Chicago and Argonne National Laboratory
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License
 */


#include <assert.h>
#include <inttypes.h>
#include <limits.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <list_i.h>
#include <list_l.h>
#include <table.h>
#include <table_bp.h>
#include <table_lp.h>
#include <vint.h>

#include "adlb.h"
#include "adlb_types.h"
#include "data.h"
#include "data_cleanup.h"
#include "data_internal.h"
#include "data_structs.h"
#include "debug.h"
#include "multiset.h"
#include "refcount.h"
#include "sync.h"

/**
   Map from adlb_datum_id to adlb_datum
*/
static struct table_lp tds;

/**
   Map from "container,subscript" specifier to list of listening references.
 */
static struct table_bp container_references;

/**
   Map from "container,subscript" specifier to list of subscribers to that
   subscript.
 */
static struct table_bp container_ix_listeners;

/**
   Map from adlb_datum_id to int rank if locked
*/
static struct table_lp locked;

/**
   Number of ADLB servers
*/
static int servers = 1;

/**
   Unique datum id.  Note that 0 is ADLB_DATA_ID_NULL.
*/
static adlb_datum_id unique = -1;

/**
   When data_unique hits this, return an error- we have exhausted the
   longs. Defined in data_init()
 */
static adlb_datum_id last_id;

static adlb_data_code
datum_init_props(adlb_datum_id id, adlb_datum *d,
                 const adlb_create_props *props);

static adlb_data_code
datum_init_container(adlb_datum *d, adlb_data_type key_type,
                 adlb_data_type val_type);

static adlb_data_code
datum_init_multiset(adlb_datum *d, adlb_data_type val_type);

static adlb_data_code
xlb_data_close(adlb_datum_id id, adlb_datum *d, int** result, int* count);
static adlb_data_code datum_gc(adlb_datum_id id, adlb_datum* d,
           refcount_scavenge scav);

static adlb_data_code
insert_notifications(adlb_datum *d,
            adlb_datum_id container_id, adlb_subscript subscript,
            adlb_datum_storage *inserted_value,
            adlb_data_type value_type,
            adlb_datums *references, adlb_ranks *notify_insert,
            bool *garbage_collected);


static bool container_lookup(adlb_container *c, adlb_subscript sub,
                             adlb_container_val *val);
static bool container_set(adlb_container *c, adlb_subscript sub,
                              adlb_container_val val,
                              adlb_container_val *prev);
static void container_add(adlb_container *c, adlb_subscript sub,
                              adlb_container_val val);

static void report_leaks(void);


// Maximum length of id/subscript string
#define ID_SUB_PAIR_MAX \
  (sizeof(adlb_datum_id) + ADLB_DATA_SUBSCRIPT_MAX + 1)

// Length of buffer for id+subscript.  Will be at most 8 bytes
// more than ADLB_SUBSCRIPT_MAX
static inline size_t id_sub_buflen(adlb_subscript sub)
{
  size_t size = (sizeof(adlb_datum_id) + sub.length);
  assert(size <= ID_SUB_PAIR_MAX);
  return size;
}

static inline size_t write_id_sub(char *buf, adlb_datum_id id,
                                  adlb_subscript sub)
{
  memcpy(buf, &id, sizeof(adlb_datum_id));
  memcpy(buf + sizeof(adlb_datum_id), sub.key, sub.length);
  return id_sub_buflen(sub);
}

// Extract id and sub from buffer.  Return internal pointer into buffer
static inline void read_id_sub(const char *buf, size_t buflen,
        adlb_datum_id *id, adlb_subscript *sub)
{
  assert(buflen >= sizeof(*id));
  memcpy(id, buf, sizeof(*id));
  sub->length = buflen - sizeof(*id);
  sub->key = &buf[sizeof(*id)];
}

/**
   @param s Number of servers
   @param server_num Number amongst servers
 */
adlb_data_code
xlb_data_init(int s, int server_num)
{
  assert(server_num >= 0 && server_num < s);
  servers = s;
  unique = server_num;
  if (unique == 0) unique += s;

  bool result;
  result = table_lp_init(&tds, 1024*1024);
  if (!result)
    return ADLB_DATA_ERROR_OOM;
  result = table_bp_init(&container_references, 1024*1024);
  result = table_bp_init(&container_ix_listeners, 1024*1024);
  if (!result)
    return ADLB_DATA_ERROR_OOM;

  result = table_lp_init(&locked, 16);
  if (!result)
    return ADLB_DATA_ERROR_OOM;

  last_id = LONG_MAX - servers - 1;

  return ADLB_DATA_SUCCESS;
}

adlb_data_code
xlb_data_create(adlb_datum_id id, adlb_data_type type,
            const adlb_type_extra *type_extra,
            const adlb_create_props *props)
{
  TRACE("data_create(%"PRId64")", id);
  check_verbose(id > 0, ADLB_DATA_ERROR_INVALID,
                "ERROR: attempt to create data: id=%"PRId64"\n", id);

  DEBUG("Create <%"PRId64"> t:%s r:%i w:%i", id, ADLB_Data_type_tostring(type),
                                props->read_refcount, props->write_refcount);
  if (type == ADLB_DATA_TYPE_CONTAINER)
    DEBUG("Create container <%"PRId64"> k:%s v:%s", id,
          ADLB_Data_type_tostring(type_extra->CONTAINER.key_type),
          ADLB_Data_type_tostring(type_extra->CONTAINER.val_type));

#ifndef NDEBUG
  check_verbose(!table_lp_contains(&tds, id), ADLB_DATA_ERROR_DOUBLE_DECLARE,
                "<%"PRId64"> already exists", id);
#endif

  if (props->read_refcount <= 0 && props->write_refcount <= 0)
  {
    DEBUG("Skipped creation of <%"PRId64">", id);
    return ADLB_DATA_SUCCESS;
  }

  adlb_datum* d = malloc(sizeof(adlb_datum));
  check_verbose(d != NULL, ADLB_DATA_ERROR_OOM,
                "Out of memory while allocating datum");
  d->type = type;
  list_i_init(&d->listeners);

  table_lp_add(&tds, id, d);
<commit_msg>Allow negative ADLB IDs for system purposes<commit_after>/*
 * Copyright 2013 University of Chicago and Argonne National Laboratory
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License
 */


#include <assert.h>
#include <inttypes.h>
#include <limits.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <list_i.h>
#include <list_l.h>
#include <table.h>
#include <table_bp.h>
#include <table_lp.h>
#include <vint.h>

#include "adlb.h"
#include "adlb_types.h"
#include "data.h"
#include "data_cleanup.h"
#include "data_internal.h"
#include "data_structs.h"
#include "debug.h"
#include "multiset.h"
#include "refcount.h"
#include "sync.h"

/**
   Map from adlb_datum_id to adlb_datum
*/
static struct table_lp tds;

/**
   Map from "container,subscript" specifier to list of listening references.
 */
static struct table_bp container_references;

/**
   Map from "container,subscript" specifier to list of subscribers to that
   subscript.
 */
static struct table_bp container_ix_listeners;

/**
   Map from adlb_datum_id to int rank if locked
*/
static struct table_lp locked;

/**
   Number of ADLB servers
*/
static int servers = 1;

/**
   Unique datum id.  Note that 0 is ADLB_DATA_ID_NULL.
*/
static adlb_datum_id unique = -1;

/**
   When data_unique hits this, return an error- we have exhausted the
   longs. Defined in data_init()
 */
static adlb_datum_id last_id;

static adlb_data_code
datum_init_props(adlb_datum_id id, adlb_datum *d,
                 const adlb_create_props *props);

static adlb_data_code
datum_init_container(adlb_datum *d, adlb_data_type key_type,
                 adlb_data_type val_type);

static adlb_data_code
datum_init_multiset(adlb_datum *d, adlb_data_type val_type);

static adlb_data_code
xlb_data_close(adlb_datum_id id, adlb_datum *d, int** result, int* count);
static adlb_data_code datum_gc(adlb_datum_id id, adlb_datum* d,
           refcount_scavenge scav);

static adlb_data_code
insert_notifications(adlb_datum *d,
            adlb_datum_id container_id, adlb_subscript subscript,
            adlb_datum_storage *inserted_value,
            adlb_data_type value_type,
            adlb_datums *references, adlb_ranks *notify_insert,
            bool *garbage_collected);


static bool container_lookup(adlb_container *c, adlb_subscript sub,
                             adlb_container_val *val);
static bool container_set(adlb_container *c, adlb_subscript sub,
                              adlb_container_val val,
                              adlb_container_val *prev);
static void container_add(adlb_container *c, adlb_subscript sub,
                              adlb_container_val val);

static void report_leaks(void);


// Maximum length of id/subscript string
#define ID_SUB_PAIR_MAX \
  (sizeof(adlb_datum_id) + ADLB_DATA_SUBSCRIPT_MAX + 1)

// Length of buffer for id+subscript.  Will be at most 8 bytes
// more than ADLB_SUBSCRIPT_MAX
static inline size_t id_sub_buflen(adlb_subscript sub)
{
  size_t size = (sizeof(adlb_datum_id) + sub.length);
  assert(size <= ID_SUB_PAIR_MAX);
  return size;
}

static inline size_t write_id_sub(char *buf, adlb_datum_id id,
                                  adlb_subscript sub)
{
  memcpy(buf, &id, sizeof(adlb_datum_id));
  memcpy(buf + sizeof(adlb_datum_id), sub.key, sub.length);
  return id_sub_buflen(sub);
}

// Extract id and sub from buffer.  Return internal pointer into buffer
static inline void read_id_sub(const char *buf, size_t buflen,
        adlb_datum_id *id, adlb_subscript *sub)
{
  assert(buflen >= sizeof(*id));
  memcpy(id, buf, sizeof(*id));
  sub->length = buflen - sizeof(*id);
  sub->key = &buf[sizeof(*id)];
}

/**
   @param s Number of servers
   @param server_num Number amongst servers
 */
adlb_data_code
xlb_data_init(int s, int server_num)
{
  assert(server_num >= 0 && server_num < s);
  servers = s;
  unique = server_num;
  if (unique == 0) unique += s;

  bool result;
  result = table_lp_init(&tds, 1024*1024);
  if (!result)
    return ADLB_DATA_ERROR_OOM;
  result = table_bp_init(&container_references, 1024*1024);
  result = table_bp_init(&container_ix_listeners, 1024*1024);
  if (!result)
    return ADLB_DATA_ERROR_OOM;

  result = table_lp_init(&locked, 16);
  if (!result)
    return ADLB_DATA_ERROR_OOM;

  last_id = LONG_MAX - servers - 1;

  return ADLB_DATA_SUCCESS;
}

adlb_data_code
xlb_data_create(adlb_datum_id id, adlb_data_type type,
            const adlb_type_extra *type_extra,
            const adlb_create_props *props)
{
  TRACE("data_create(%"PRId64")", id);
  check_verbose(id != ADLB_DATA_ID_NULL, ADLB_DATA_ERROR_INVALID,
                "ERROR: attempt to create data: id=%"PRId64"\n", id);

  DEBUG("Create <%"PRId64"> t:%s r:%i w:%i", id, ADLB_Data_type_tostring(type),
                                props->read_refcount, props->write_refcount);
  if (type == ADLB_DATA_TYPE_CONTAINER)
    DEBUG("Create container <%"PRId64"> k:%s v:%s", id,
          ADLB_Data_type_tostring(type_extra->CONTAINER.key_type),
          ADLB_Data_type_tostring(type_extra->CONTAINER.val_type));

#ifndef NDEBUG
  check_verbose(!table_lp_contains(&tds, id), ADLB_DATA_ERROR_DOUBLE_DECLARE,
                "<%"PRId64"> already exists", id);
#endif

  if (props->read_refcount <= 0 && props->write_refcount <= 0)
  {
    DEBUG("Skipped creation of <%"PRId64">", id);
    return ADLB_DATA_SUCCESS;
  }

  adlb_datum* d = malloc(sizeof(adlb_datum));
  check_verbose(d != NULL, ADLB_DATA_ERROR_OOM,
                "Out of memory while allocating datum");
  d->type = type;
  list_i_init(&d->listeners);

  table_lp_add(&tds, id, d);

{'fraction_non_alphanumeric': 0.10240223005657129, 'fraction_numerical': 0.00926457325571862, 'mean_word_length': 3.70420362514462, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 62, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4689939', 'n_tokens_mistral': 4417, 'n_tokens_neox': 4220, 'n_words': 1086}