<commit_before>/* Copyright 2016 Stanford University, NVIDIA Corporation
 *
 * 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 "activemsg.h"
#include "utilities.h"

#ifndef __GNUC__
#include "atomics.h" // for __sync_add_and_fetch
#endif

#include <queue>
#include <cassert>
#ifdef REALM_PROFILE_AM_HANDLERS
#include <math.h>
#endif

#include "lowlevel_impl.h"

#include "realm/timers.h"
#include "realm/logging.h"

#define NO_DEBUG_AMREQUESTS

enum { MSGID_LONG_EXTENSION = 253,
       MSGID_FLIP_REQ = 254,
       MSGID_FLIP_ACK = 255 };

#ifdef DEBUG_MEM_REUSE
static int payload_count = 0;
#endif

LegionRuntime::Logger::Category log_amsg("activemsg");
LegionRuntime::Logger::Category log_spill("spill");

#ifdef ACTIVE_MESSAGE_TRACE
LegionRuntime::Logger::Category log_amsg_trace("amtrace");

void record_am_handler(int handler_id, const char *description, bool reply)
{
  log_amsg_trace.info("AM Handler: %d %s %s\n", handler_id, description,
		      (reply ? "Reply" : "Request"));
}
#endif

#ifdef REALM_PROFILE_AM_HANDLERS
/*extern*/ ActiveMsgHandlerStats handler_stats[256];
#endif

static const int DEFERRED_FREE_COUNT = 128;
gasnet_hsl_t deferred_free_mutex;
int deferred_free_pos;
void *deferred_frees[DEFERRED_FREE_COUNT];

gasnet_seginfo_t *segment_info = 0;

static bool is_registered(void *ptr)
{
  ssize_t offset = ((char *)ptr) - ((char *)(segment_info[gasnet_mynode()].addr));
  if((offset >= 0) && ((size_t)offset < segment_info[gasnet_mynode()].size))
    return true;
  return false;
}

class IncomingMessageManager {
public:
  IncomingMessageManager(int _nodes, Realm::CoreReservationSet& crs);
  ~IncomingMessageManager(void);

  void add_incoming_message(int sender, IncomingMessage *msg);

  void start_handler_threads(int count, size_t stack_size);

  void shutdown(void);

  IncomingMessage *get_messages(int &sender, bool wait = true);

  void handler_thread_loop(void);

protected:
  int nodes;
  int shutdown_flag;
  IncomingMessage **heads;
  IncomingMessage ***tails;
  int *todo_list; // list of nodes with non-empty message lists
  int todo_oldest, todo_newest;
  gasnet_hsl_t mutex;
  gasnett_cond_t condvar;
  Realm::CoreReservation *core_rsrv;
  std::vector<Realm::Thread *> handler_threads;
};

void init_deferred_frees(void)
{
  gasnet_hsl_init(&deferred_free_mutex);
  deferred_free_pos = 0;
  for(int i = 0; i < DEFERRED_FREE_COUNT; i++)
    deferred_frees[i] = 0;
}

void deferred_free(void *ptr)
{
#ifdef DEBUG_MEM_REUSE
  printf("%d: deferring free of %p\n", gasnet_mynode(), ptr);
#endif
  gasnet_hsl_lock(&deferred_free_mutex);
  void *oldptr = deferred_frees[deferred_free_pos];
  deferred_frees[deferred_free_pos] = ptr;
  deferred_free_pos = (deferred_free_pos + 1) % DEFERRED_FREE_COUNT;
  gasnet_hsl_unlock(&deferred_free_mutex);
  if(oldptr) {
#ifdef DEBUG_MEM_REUSE
    printf("%d: actual free of %p\n", gasnet_mynode(), oldptr);
#endif
    free(oldptr);
  }
}

class PayloadSource {
public:
  PayloadSource(void) { }
  virtual ~PayloadSource(void) { }
public:
  virtual void copy_data(void *dest) = 0;
  virtual void *get_contig_pointer(void) { assert(false); return 0; }
  virtual int get_payload_mode(void) { assert(false); return PAYLOAD_NONE; }
};

class ContiguousPayload : public PayloadSource {
public:
  ContiguousPayload(void *_srcptr, size_t _size, int _mode);
  virtual ~ContiguousPayload(void) { }
  virtual void copy_data(void *dest);
  virtual void *get_contig_pointer(void) { return srcptr; }
  virtual int get_payload_mode(void) { return mode; }
protected:
  void *srcptr;
  size_t size;
  int mode;
};

class TwoDPayload : public PayloadSource {
public:
  TwoDPayload(const void *_srcptr, size_t _line_size, size_t _line_count,
	      ptrdiff_t _line_stride, int _mode);
  virtual ~TwoDPayload(void) { }
  virtual void copy_data(void *dest);
protected:
  const void *srcptr;
  size_t line_size, line_count;
  ptrdiff_t line_stride;
  int mode;
};<commit_msg>realm: Temporary fix in the active message handler<commit_after>/* Copyright 2016 Stanford University, NVIDIA Corporation
 *
 * 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 "activemsg.h"
#include "utilities.h"

#ifndef __GNUC__
#include "atomics.h" // for __sync_add_and_fetch
#endif

#include <queue>
#include <cassert>
#ifdef REALM_PROFILE_AM_HANDLERS
#include <math.h>
#endif

#include "lowlevel_impl.h"

#include "realm/timers.h"
#include "realm/logging.h"

#define NO_DEBUG_AMREQUESTS

enum { MSGID_LONG_EXTENSION = 253,
       MSGID_FLIP_REQ = 254,
       MSGID_FLIP_ACK = 255 };

#ifdef DEBUG_MEM_REUSE
static int payload_count = 0;
#endif

LegionRuntime::Logger::Category log_amsg("activemsg");
LegionRuntime::Logger::Category log_spill("spill");

#ifdef ACTIVE_MESSAGE_TRACE
LegionRuntime::Logger::Category log_amsg_trace("amtrace");

void record_am_handler(int handler_id, const char *description, bool reply)
{
  log_amsg_trace.info("AM Handler: %d %s %s\n", handler_id, description,
		      (reply ? "Reply" : "Request"));
}
#endif

#ifdef REALM_PROFILE_AM_HANDLERS
/*extern*/ ActiveMsgHandlerStats handler_stats[256];
#endif

static const int DEFERRED_FREE_COUNT = 128;
gasnet_hsl_t deferred_free_mutex;
int deferred_free_pos;
void *deferred_frees[DEFERRED_FREE_COUNT];

gasnet_seginfo_t *segment_info = 0;

static bool is_registered(void *ptr)
{
  ssize_t offset = ((char *)ptr) - ((char *)(segment_info[gasnet_mynode()].addr));
  if((offset >= 0) && ((size_t)offset < segment_info[gasnet_mynode()].size))
    return true;
  return false;
}

class IncomingMessageManager {
public:
  IncomingMessageManager(int _nodes, Realm::CoreReservationSet& crs);
  ~IncomingMessageManager(void);

  void add_incoming_message(int sender, IncomingMessage *msg);

  void start_handler_threads(int count, size_t stack_size);

  void shutdown(void);

  IncomingMessage *get_messages(int &sender, bool wait = true);

  void handler_thread_loop(void);

protected:
  int nodes;
  int shutdown_flag;
  IncomingMessage **heads;
  IncomingMessage ***tails;
  int *todo_list; // list of nodes with non-empty message lists
  int todo_oldest, todo_newest;
  gasnet_hsl_t mutex;
  gasnett_cond_t condvar;
  Realm::CoreReservation *core_rsrv;
  std::vector<Realm::Thread *> handler_threads;
};

void init_deferred_frees(void)
{
  gasnet_hsl_init(&deferred_free_mutex);
  deferred_free_pos = 0;
  for(int i = 0; i < DEFERRED_FREE_COUNT; i++)
    deferred_frees[i] = 0;
}

void deferred_free(void *ptr)
{
#ifdef DEBUG_MEM_REUSE
  printf("%d: deferring free of %p\n", gasnet_mynode(), ptr);
#endif
  gasnet_hsl_lock(&deferred_free_mutex);
  void *oldptr = deferred_frees[deferred_free_pos];
  deferred_frees[deferred_free_pos] = ptr;
  deferred_free_pos = (deferred_free_pos + 1) % DEFERRED_FREE_COUNT;
  gasnet_hsl_unlock(&deferred_free_mutex);
  if(oldptr) {
#ifdef DEBUG_MEM_REUSE
    printf("%d: actual free of %p\n", gasnet_mynode(), oldptr);
#endif
    free(oldptr);
  }
}

class PayloadSource {
public:
  PayloadSource(void) { }
  virtual ~PayloadSource(void) { }
public:
  virtual void copy_data(void *dest) = 0;
  virtual void *get_contig_pointer(void) { assert(false); return 0; }
  virtual int get_payload_mode(void) { return PAYLOAD_KEEP; }
};

class ContiguousPayload : public PayloadSource {
public:
  ContiguousPayload(void *_srcptr, size_t _size, int _mode);
  virtual ~ContiguousPayload(void) { }
  virtual void copy_data(void *dest);
  virtual void *get_contig_pointer(void) { return srcptr; }
  virtual int get_payload_mode(void) { return mode; }
protected:
  void *srcptr;
  size_t size;
  int mode;
};

class TwoDPayload : public PayloadSource {
public:
  TwoDPayload(const void *_srcptr, size_t _line_size, size_t _line_count,
	      ptrdiff_t _line_stride, int _mode);
  virtual ~TwoDPayload(void) { }
  virtual void copy_data(void *dest);
protected:
  const void *srcptr;
  size_t line_size, line_count;
  ptrdiff_t line_stride;
  int mode;
};
{'fraction_non_alphanumeric': 0.10695067264573992, 'fraction_numerical': 0.007174887892376682, 'mean_word_length': 5.376697641172266, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5176313', 'n_tokens_mistral': 3186, 'n_tokens_neox': 3081, 'n_words': 784}