mop3/kernel/proc/mail.c

#include <libk/list.h>
#include <libk/minmax.h>
#include <libk/std.h>
#include <libk/string.h>
#include <mm/liballoc.h>
#include <proc/mail.h>
#include <proc/proc.h>
#include <proc/resource.h>
#include <proc/suspension_q.h>
#include <sys/debug.h>
#include <sys/smp.h>

void proc_cleanup_resource_mail (struct proc_resource* resource, struct reschedule_ctx* rctx) {
  struct proc_mail* mail = &resource->u.mail;

  spin_lock (&mail->resource->lock);

  if (mail->pending_mesg != NULL)
    free (mail->pending_mesg);

  mail->pending_mesg_size = 0;

  spin_lock (&mail->send_sq.lock);

  while (mail->send_sq.proc_list != NULL) {
    struct list_node_link* node = mail->send_sq.proc_list;
    struct proc_sq_entry* sq_entry = list_entry (node, struct proc_sq_entry, sq_link);
    struct proc* suspended_proc = sq_entry->proc;

    spin_unlock (&mail->send_sq.lock);
    spin_unlock (&mail->resource->lock);

    proc_sq_resume (suspended_proc, sq_entry, rctx);

    spin_lock (&mail->resource->lock);
    spin_lock (&mail->send_sq.lock);
  }

  spin_unlock (&mail->send_sq.lock);
  spin_unlock (&mail->resource->lock);
}

void proc_mail_send (struct proc* proc, struct proc_mail* mail, struct reschedule_ctx* rctx,
                     void* data, size_t data_size) {
  spin_lock (&mail->resource->lock);

  /* mail full */
  if (mail->pending_mesg != NULL) {
    proc_sq_suspend (proc, &mail->send_sq, &mail->resource->lock, rctx);
    return;
  }

  spin_lock (&mail->recv_sq.lock);

  /* if receiver available, hand off directly */
  struct list_node_link* node = mail->recv_sq.proc_list;

  if (node != NULL) {
    struct proc_sq_entry* sq_entry = list_entry (node, struct proc_sq_entry, sq_link);
    struct proc* resumed_proc = sq_entry->proc;

    spin_unlock (&mail->recv_sq.lock);
    spin_unlock (&mail->resource->lock);

    spin_lock (&resumed_proc->lock);

    if (resumed_proc->mail_recv_buffer != NULL) {
      size_t copy_size = min (data_size, resumed_proc->mail_recv_size);
      memcpy (resumed_proc->mail_recv_buffer, data, copy_size);

      resumed_proc->mail_recv_buffer = NULL;
      resumed_proc->mail_recv_size = 0;
    }

    spin_unlock (&resumed_proc->lock);

    proc_sq_resume (resumed_proc, sq_entry, rctx);
    return;
  }

  spin_unlock (&mail->recv_sq.lock);

  /* mail is empty and nobody is waiting */
  void* mesg = malloc (data_size);
  if (mesg != NULL) {
    mail->pending_mesg = mesg;
    memcpy (mail->pending_mesg, data, data_size);
    mail->pending_mesg_size = data_size;
  }

  spin_unlock (&mail->resource->lock);
}

void proc_mail_receive (struct proc* proc, struct proc_mail* mail, struct reschedule_ctx* rctx,
                        void* recv_buffer, size_t recv_size) {
  spin_lock (&proc->lock);
  proc->mail_recv_buffer = recv_buffer;
  proc->mail_recv_size = recv_size;
  spin_unlock (&proc->lock);

  spin_lock (&mail->resource->lock);

  /* consume mesg if available */
  if (mail->pending_mesg != NULL) {
    memcpy (recv_buffer, mail->pending_mesg, min (recv_size, mail->pending_mesg_size));
    free (mail->pending_mesg);
    mail->pending_mesg = NULL;
    mail->pending_mesg_size = 0;

    /* check for suspended sender */
    spin_lock (&mail->send_sq.lock);

    struct list_node_link* node = mail->send_sq.proc_list;

    if (node != NULL) {
      struct proc_sq_entry* sq_entry = list_entry (node, struct proc_sq_entry, sq_link);
      struct proc* resumed_proc = sq_entry->proc;

      spin_unlock (&mail->send_sq.lock);
      spin_unlock (&mail->resource->lock);

      proc_sq_resume (resumed_proc, sq_entry, rctx);
      return;
    }

    spin_unlock (&mail->send_sq.lock);
    spin_unlock (&mail->resource->lock);

    return;
  }

  /* nothing to receive */
  proc_sq_suspend (proc, &mail->recv_sq, &mail->resource->lock, rctx);
}