/* messenger.c - Messenger object implementation.
   Copyright (C) 2008 Free Software Foundation, Inc.
   Written by Neal H. Walfield <neal@gnu.org>.

   This file is part of the GNU Hurd.

   The GNU Hurd is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License as
   published by the Free Software Foundation; either version 3 of the
   License, or (at your option) any later version.

   The GNU Hurd is distributed in the hope that it will be useful, but
   WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program.  If not, see
   <http://www.gnu.org/licenses/>.  */

#include <stdbool.h>
#include <errno.h>
#include <assert.h>
#include <viengoos/cap.h>
#include <viengoos/misc.h>
#include <hurd/as.h>

#include "messenger.h"
#include "object.h"
#include "thread.h"
#include "timer.h"
#include "activity.h"

/* When the kernel formulates relies, it does so in this buffer.  */
static char reply_message_data[PAGESIZE] __attribute__ ((aligned (PAGESIZE)));
struct vg_message *reply_buffer = (struct vg_message *) &reply_message_data[0];

#define MIN(x, y) ((x) < (y) ? (x) : (y))
#define MAX(x, y) ((x) > (y) ? (x) : (y))

#include <backtrace.h>

static bool
messenger_load_internal (struct activity *activity,
			 struct messenger *target,
			 struct messenger *source,
			 struct vg_message *smessage,
			 bool may_block)
{
  assert (object_type ((struct vg_object *) target) == vg_cap_messenger);
  if (source)
    assert (object_type ((struct vg_object *) source) == vg_cap_messenger);

  if (source)
    assert (! smessage);
  else
    assert (smessage);

  /* SOURCE should not already be blocked on another messenger.  */
  if (source)
    {
      assert (! source->wait_queue.next);
      assert (! source->wait_queue.prev);
    }

  if (unlikely (target->blocked))
    /* TARGET is blocked.  */
    {
      if (! may_block)
	{
	  debug (0, "Not enqueuing messenger "VG_OID_FMT" on "VG_OID_FMT": "
		 "target blocked and delivery marked as non-blocking "
		 "(target->thread: "VG_OID_FMT").",
		 source ? object_oid ((struct vg_object *) source) : 0,
		 object_oid ((struct vg_object *) target),
		 target->thread.oid);
	  backtrace_print ();
	  return false;
	}

      /* Enqueue SOURCE on TARGET's wait queue.  */

      debug (0, "Target blocked.  Enqueuing sender.");

      assert (source);
      source->wait_reason = MESSENGER_WAIT_TRANSFER_MESSAGE;
      object_wait_queue_enqueue (activity, (struct vg_object *) target, source);

      return true;
    }

  /* TARGET is not blocked.  Deliver the message.  */
  debug (5, "Delivering sender's message to target.");

  target->blocked = true;

  /* There are four combinations: the source can either have inline
     data or out-of-line data and the target can either have inline
     data or out-of-line data.  */

  struct vg_message *tmessage = NULL;

  void *sdata;
  void *tdata;
  int data_count;

  vg_addr_t *saddrs;
  int saddr_count;
  vg_addr_t *taddrs;
  int taddr_count;

  if (! source || source->out_of_band)
    /* Source data is in a buffer.  */
    {
      if (source)
	smessage = (struct vg_message *) vg_cap_to_object (activity,
							&source->buffer);
      else
	assert (smessage);

      if (smessage)
	{
	  sdata = vg_message_data (smessage);
	  data_count = vg_message_data_count (smessage);

	  saddrs = vg_message_caps (smessage);
	  saddr_count = vg_message_cap_count (smessage);
	}
      else
	{
	  sdata = NULL;
	  data_count = 0;
	  saddrs = NULL;
	  saddr_count = 0;
	}
    }
  else
    /* Source data is inline.  */
    {
      assert (source);

      sdata = source->inline_words;
      data_count
	= sizeof (source->inline_words[0]) * source->inline_word_count;

      saddrs = source->inline_caps;
      saddr_count = source->inline_cap_count;
    }

  if (target->out_of_band)
    /* Target data is in a buffer.  */
    {
      tmessage = (struct vg_message *) vg_cap_to_object (activity,
						      &target->buffer);
      if (tmessage)
	{
	  taddrs = vg_message_caps (tmessage);
	  taddr_count = vg_message_cap_count (tmessage);

	  /* Set the number of capabilities to the number in the
	     source message.  */
	  tmessage->cap_count = saddr_count;
	  tdata = vg_message_data (tmessage);
	  tmessage->data_count = data_count;
	}
      else
	{
	  tdata = NULL;
	  data_count = 0;

	  taddrs = NULL;
	  taddr_count = 0;
	}
    }
  else
    /* Target data is inline.  */
    {
      tdata = target->inline_words;
      data_count = MIN (data_count,
			sizeof (uintptr_t) * VG_MESSENGER_INLINE_WORDS);
      target->inline_word_count
	= (data_count + sizeof (uintptr_t) - 1) / sizeof (uintptr_t);

      taddrs = target->inline_caps;
      taddr_count = target->inline_cap_count;
    }

  do_debug (5)
    {
      if (smessage)
	{
	  debug (0, "Source: ");
	  vg_message_dump (smessage);
	}
      if (tmessage)
	{
	  debug (0, "Target: ");
	  vg_message_dump (tmessage);
	}
    }

  /* Copy the caps.  */
  int i;
  for (i = 0; i < MIN (saddr_count, taddr_count); i ++)
    {
      /* First get the target capability slot.  */
      bool twritable = true;

      struct vg_cap *tcap = NULL;
      if (! VG_ADDR_IS_VOID (taddrs[i]))
	{
	  as_slot_lookup_rel_use (activity, &target->as_root, taddrs[i],
				  ({
				    twritable = writable;
				    tcap = slot;
				  }));
	  if (! tcap || ! twritable)
	    debug (0, DEBUG_BOLD ("Target " VG_ADDR_FMT " does not designate "
				  "a %svalid slot!"),
		   VG_ADDR_PRINTF (taddrs[i]), twritable ? "writable " : "");
	}

      if (likely (tcap && twritable))
	/* We have a slot and it is writable.  Look up the source
	   capability.  */
	{
	  struct vg_cap scap = VG_CAP_VOID;
	  bool swritable = true;
	  if (source)
	    {
	      if (! VG_ADDR_IS_VOID (saddrs[i]))
		scap = as_cap_lookup_rel (activity,
					  &source->as_root, saddrs[i],
					  -1, &swritable);
	    }
	  else
	    /* This is a kernel provided buffer.  In this case the
	       address is really a pointer to a capability.  */
	    if ((uintptr_t) saddrs[i].raw)
	      scap = * (struct vg_cap *) (uintptr_t) saddrs[i].raw;

	  if (! swritable)
	    scap.type = vg_cap_type_weaken (scap.type);

	  /* Shoot down the capability.  */
	  cap_shootdown (activity, tcap);

	  /* Preserve the address translator and policy.  */
	  struct vg_cap_properties props = VG_CAP_PROPERTIES_GET (*tcap);
	  *tcap = scap;
	  VG_CAP_PROPERTIES_SET (tcap, props);

	  debug (5, VG_ADDR_FMT " <- " VG_CAP_FMT,
		 VG_ADDR_PRINTF (taddrs[i]), VG_CAP_PRINTF (tcap));
	}
      else
	taddrs[i] = VG_ADDR_VOID;
    }
  if (i < MAX (taddr_count, saddr_count) && target->out_of_band && taddrs)
    /* Set the address of any non-transferred caps in the target to
       VG_ADDR_VOID.  */
    memset (&taddrs[i], 0,
	    sizeof (taddrs[0]) * (MAX (taddr_count, saddr_count)) - i);

  /* Copy the data.  */
  memcpy (tdata, sdata, data_count);

  do_debug (5)
    if (tmessage)
      {
	debug (0, "Delivery: ");
	vg_message_dump (tmessage);
      }

  if (target->activate_on_receive)
    messenger_message_deliver (activity, target);
  else
    debug (0, "Not activing target.");

  if (source && source->activate_on_send)
    messenger_message_deliver (activity, source);

  return true;
}

bool
messenger_message_transfer (struct activity *activity,
			    struct messenger *target,
			    struct messenger *source,
			    bool may_block)
{
  return messenger_load_internal (activity, target, source, NULL, may_block);
}

bool
messenger_message_load (struct activity *activity,
			struct messenger *target,
			struct vg_message *message)
{
  return messenger_load_internal (activity, target, NULL, message, false);
}

bool
messenger_message_deliver (struct activity *activity,
			   struct messenger *messenger)
{
  assert (messenger->blocked);
  assertx (! messenger->wait_queue_p,
	   VG_OID_FMT" -> (h:%d)"VG_OID_FMT"(t:%d), reason: %d -> "VG_OID_FMT,
	   messenger->wait_queue.prev,
	   messenger->wait_queue_head,
	   object_oid ((struct vg_object *) messenger),
	   messenger->wait_queue_tail,
	   messenger->wait_reason,
	   messenger->wait_queue.next);

  struct thread *thread
    = (struct thread *) vg_cap_to_object (activity, &messenger->thread);
  if (! thread)
    {
      debug (0, "Messenger has no thread to activate!");
      return false;
    }

  if (object_type ((struct vg_object *) thread) != vg_cap_thread)
    {
      debug (0, "Messenger's thread vg_cap does not designate a thread but a %s",
	     vg_cap_type_string (object_type ((struct vg_object *) thread)));
      return false;
    }

  return thread_activate (activity, thread, messenger, true);
}

void
messenger_unblock (struct activity *activity, struct messenger *messenger)
{
  if (! messenger->blocked)
    return;

  messenger->blocked = 0;

  struct messenger *m;
  object_wait_queue_for_each (activity, (struct vg_object *) messenger, m)
    if (m->wait_reason == MESSENGER_WAIT_TRANSFER_MESSAGE)
      {
	object_wait_queue_unlink (activity, m);
	bool ret = messenger_message_transfer (activity, messenger, m, true);
	assert (ret);

	break;
      }
}

struct vg_object *sleep_object;

static struct timer sleep_timer;
static bool sleep_timer_registered;

#define TIME_FMT "%"PRId64".%03"PRId64
#define TIME_PRINTF(t)				\
  ((t) / 1000 / 1000 / 1000),			\
  (((t) / 1000 / 1000) % 1000)

static void
wake_up (struct timer *timer)
{
  int d = 5;
  debug (d, "Bzzz. Bzzz.");

  assert (&sleep_timer == timer);
  assert (sleep_timer_registered);

  struct messenger *m;
  while ((m = object_wait_queue_head (root_activity, sleep_object)))
    {
      debug (d, "Considering waking messenger "VG_OID_FMT". "
	     "now: "TIME_FMT"; wake: "TIME_FMT,
	     object_oid ((struct vg_object *) m),
	     TIME_PRINTF (time_data.ns_since_boot),
	     TIME_PRINTF (m->wait_reason_raw));

      assert (m->wait_reason == MESSENGER_SLEEP);
      if (m->wait_reason_raw <= time_data.ns_since_boot)
	{
	  object_wait_queue_unlink (root_activity, m);
	  vg_sleep_reply (root_activity, m);
	}
      else
	break;
    }

  if (m)
    {
      debug (d, "Registering timer callback at "TIME_FMT".",
	     TIME_PRINTF (m->wait_reason_raw));
      sleep_timer.expire = m->wait_reason_raw;
      timer_register (&sleep_timer);
    }
  else
    sleep_timer_registered = false;
}

/* Enqueue the messenger MESSENGER on SLEEP_OBJECT and deliver it
   after TIMEOUT nanoseconds.  */
void
messenger_sleep (struct activity *activity,
		 struct messenger *messenger,
		 uint64_t timeout)
{
  int d = 5;

  uint64_t deadline = time_data.ns_since_boot + timeout;

  debug (d, VG_OID_FMT" to sleep for "TIME_FMT" seconds. "
	 "(now: "TIME_FMT"; wake: "TIME_FMT")",
	 object_oid ((struct vg_object *) messenger), TIME_PRINTF (timeout),
	 TIME_PRINTF (time_data.ns_since_boot), TIME_PRINTF (deadline));

  messenger->wait_reason = MESSENGER_SLEEP;
  messenger->wait_reason_raw = deadline;

  struct messenger *last = NULL;
  struct messenger *m;
  object_wait_queue_for_each (activity, sleep_object, m)
    {
      assert (m->wait_reason == MESSENGER_SLEEP);
      if (deadline < m->wait_reason_raw)
	break;

      last = m;
    }

  if (last)
    {
      object_wait_queue_insert_after (activity, sleep_object, last, messenger);
      assert (sleep_timer_registered);
    }
  else
    {
      debug (d, "Adding messenger "VG_OID_FMT" to head of sleep queue",
	     object_oid ((struct vg_object *) messenger));
      object_wait_queue_push (activity, sleep_object, messenger);

      if (sleep_timer_registered)
	{
	  debug (d, "Cancelling pending timer.");
	  timer_cancel (&sleep_timer);
	}

      sleep_timer.expire = deadline;
      sleep_timer.callback = wake_up;
      sleep_timer_registered = true;
      timer_register (&sleep_timer);
    }
}

void
messenger_destroy (struct activity *activity, struct messenger *messenger)
{
  if (messenger->wait_queue_p)
    /* MESSENGER is attached to a wait queue.  Detach it.  */
    object_wait_queue_unlink (activity, messenger);
}