loop.c File Reference

a simple object-oriented event loop More...

#include <stdlib.h>
#include <stddef.h>
#include <stdbool.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <sys/epoll.h>
#include <signal.h>
#include <string.h>
#include <time.h>
#include "debug.h"
#include "process.h"
#include "socket.h"
#include "loop.h"
#include "list.h"

Functions
int	co_loop_create (void)
	sets up the event loop
int	co_loop_destroy (void)
	closes the event loop
void	co_loop_start (void)
	starts the event loop, listening for messages on sockets via _co_loop_poll_sockets
void	co_loop_stop (void)
	stops the event loop
int	co_loop_add_process (co_obj_t *proc)
	adds a process to the event loop (for it to listen for) More...
int	co_loop_remove_process (pid_t pid)
	removes a process from the event loop More...
int	co_loop_add_socket (co_obj_t *new_sock, co_obj_t *context)
	adds a new socket to the event loop (for it to listen on) More...
int	co_loop_remove_socket (co_obj_t *old_sock, co_obj_t *context)
	removes a socket from the event loop More...
co_obj_t *	co_loop_get_socket (char *uri, co_obj_t *context)
	gets a socket that is registered with the event loop More...
int	co_loop_add_timer (co_obj_t *new_timer, co_obj_t *context)
	schedules a new timer with the event loop More...
int	co_loop_remove_timer (co_obj_t *old_timer, co_obj_t *context)
	removes a timer from the event loop More...
co_obj_t *	co_loop_get_timer (void *ptr, co_obj_t *context)
	gets a socket that is registered with the event loop More...
int	co_loop_set_timer (co_obj_t *old_timer, long msecs, co_obj_t *context)
	sets timer to expire in msecs from now More...
co_obj_t *	co_timer_create (struct timeval deadline, co_cb_t timer_cb, void *ptr)
	malloc and initialize a timer More...

Detailed Description

a simple object-oriented event loop

Author

Josh King (jheretic), jking.nosp@m.@cha.nosp@m.mbana.nosp@m..net

Function Documentation

int co_loop_add_process

(

co_obj_t *

proc

)

adds a process to the event loop (for it to listen for)

Parameters

proc	the process to be added

References co_list_append().

                                        {
  CHECK(IS_PROCESS(proc),"Not a process.");
  CHECK(co_list_append(processes,proc),"Failed to add process %d",((co_process_t*)proc)->pid);
  ((co_process_t*)proc)->registered = true; 
  return 1;
error:
  return 0;
}

int co_loop_add_socket	(	co_obj_t *	new_sock,
		co_obj_t *	context
	)

adds a new socket to the event loop (for it to listen on)

Parameters

new_sock	the new socket to be added
context	a co_obj_t context pointer (currently unused)

References co_list_append(), co_list_contains(), co_list_parse(), and co_loop_remove_socket().

Referenced by _watch(), and main().

                                                              {
  DEBUG("Adding socket to event loop.");
  CHECK(IS_SOCK(new_sock),"Not a socket.");
  co_socket_t *sock = (co_socket_t*)new_sock;
  co_fd_t *fd = (co_fd_t*)context;
  CHECK(fd->socket == sock,"FD does not match socket");
  co_obj_t *node = NULL;
  struct epoll_event event;

  memset(&event, 0, sizeof(struct epoll_event));
  event.events = EPOLLIN;

  if((node = co_list_parse(sockets, _co_loop_match_socket_i, sock->uri))) {
    CHECK((node == (co_obj_t*)sock), "Different socket with URI %s already registered.", sock->uri);
    if ((sock->listen) && (sock->fd->fd > 0) && sock->fd_registered) {
      CHECK(co_list_contains(sock->rfd_lst,(co_obj_t*)fd),"Socket does not contain FD");
      DEBUG("Adding RFD %d to epoll.", fd->fd);
      event.data.ptr = (co_obj_t*)fd;
      CHECK((epoll_ctl(poll_fd, EPOLL_CTL_ADD, fd->fd, &event)) != -1, "Failed to add receive FD epoll event.");
    } else {
      SENTINEL("Socket %s already registered.", sock->uri);
    }
  } else if((sock->listen) && (sock->fd->fd > 0) && !sock->fd_registered) {
      CHECK(fd = sock->fd,"Invalid listening socket");
      DEBUG("Adding FD %d to epoll.", sock->fd->fd);
      event.data.ptr = (co_obj_t*)fd;
      CHECK((epoll_ctl(poll_fd, EPOLL_CTL_ADD, sock->fd->fd, &event)) != -1, "Failed to add listen FD epoll event.");
      sock->fd_registered = true; 
      co_list_append(sockets, (co_obj_t*)sock);
      return 1;
  } else {
      co_loop_remove_socket((co_obj_t*)sock, NULL);
      SENTINEL("Unknown error registering socket %s.", sock->uri);
  }

error:
  return 0;
}

int co_loop_add_timer	(	co_obj_t *	new_timer,
		co_obj_t *	context
	)

schedules a new timer with the event loop

Parameters

timer	the timer to schedule
context	a co_obj_t context pointer (currently unused)

References co_list_append(), co_list_insert_before(), and co_list_parse().

Referenced by _schedule(), and co_loop_set_timer().

                                                              {
  co_timer_t *timer = (co_timer_t*)new_timer;
  co_obj_t *node = NULL;
  struct timeval now;
  
  if (timer->pending)
    return 0;
  
  _co_loop_gettime(&now);
  CHECK(timer->timer_cb,"No callback function associated with timer");
  CHECK(_co_loop_tv_diff(&timer->deadline,&now) > -1000,"Invalid timer deadline: %ld.%06ld  %ld.%06ld",
  timer->deadline.tv_sec,timer->deadline.tv_usec,
  now.tv_sec,now.tv_usec);
    
  CHECK(co_list_parse(timers,_co_loop_match_timer_i,timer->ptr) == NULL,"Timer already scheduled");
  
  // insert into list in chronological order
  if((node = co_list_parse(timers, _co_loop_compare_timer_i, &timer->deadline))) {
    CHECK(co_list_insert_before(timers,(co_obj_t*)timer,node),"Failed to insert timer.");
  } else {
    CHECK(co_list_append(timers,(co_obj_t*)timer),"Failed to insert timer.");
  }
  
//   DEBUG("Successfully added timer %ld.%06ld %p",timer->deadline.tv_sec,timer->deadline.tv_usec,timer->ptr);
  
  timer->pending = true;
  return 1;
  
error:
  return 0;
}

co_obj_t* co_loop_get_socket	(	char *	uri,
		co_obj_t *	context
	)

gets a socket that is registered with the event loop

Parameters

uri	a URI to match against the available sockets
context	a co_obj_t context pointer (currently unused)

References co_list_parse().

                                                           {
  co_obj_t *sock = NULL;
  CHECK((sock = co_list_parse(sockets, _co_loop_match_socket_i, uri)), "Failed to find socket %s", uri);
  return sock;
error:
  return NULL;
}

co_obj_t* co_loop_get_timer	(	void *	ptr,
		co_obj_t *	context
	)

gets a socket that is registered with the event loop

Parameters

ptr	void pointer to match against the available timers
context	a co_obj_t context pointer (currently unused)

References co_list_parse().

Referenced by _unschedule().

                                                          {
  co_obj_t *timer = NULL;
  CHECK((timer = co_list_parse(timers, _co_loop_match_timer_i, ptr)), "Failed to find timer: %p",ptr);
  return timer;
error:
  return NULL;
}

int co_loop_remove_process

(

pid_t

pid

)

removes a process from the event loop

Parameters

pid	the id of the process to be removed

References co_list_delete(), and co_list_parse().

                                      {
  co_obj_t *proc = NULL;
  CHECK((proc = co_list_parse(processes, _co_loop_match_process_i, &pid)), "Failed to delete process %d!", pid);
  proc = co_list_delete(processes, proc);
  ((co_process_t*)proc)->registered = false; 
  return 1;

error:
  return 0;
}

int co_loop_remove_socket	(	co_obj_t *	old_sock,
		co_obj_t *	context
	)

removes a socket from the event loop

Parameters

old_sock	the socket to be removed
context	a co_obj_t context pointer (currently unused)

References co_list_delete(), and co_list_parse().

Referenced by co_loop_add_socket().

                                                                 {
  co_socket_t *sock = (co_socket_t*)old_sock;
  co_obj_t *node = NULL;
  CHECK((node = co_list_parse(sockets, _co_loop_match_socket_i, sock->uri)), "Failed to delete socket %s!", sock->uri);
  co_list_delete(sockets, node);
  sock->fd_registered = false; 
  epoll_ctl(poll_fd, EPOLL_CTL_DEL, sock->fd->fd, NULL);
  CHECK(co_list_parse(sock->rfd_lst,_co_loop_remove_fd_i,NULL) == NULL,"Failed to delete rfd_lst");
  return 1;

error:
  return 0;
}

int co_loop_remove_timer	(	co_obj_t *	old_timer,
		co_obj_t *	context
	)

removes a timer from the event loop

Parameters

old_timer	the timer to remove from list
context	a co_obj_t context pointer (currently unused)

References co_list_delete(), and co_list_parse().

Referenced by _unschedule(), and co_loop_set_timer().

                                                                 {
  co_timer_t *timer = (co_timer_t*)old_timer;
  co_obj_t *node = NULL;
  
  if (!timer->pending)
    return 0;
  
  // remove from list
  CHECK((node = co_list_parse(timers, _co_loop_match_timer_i, timer->ptr)),
  "Failed to delete timer %ld.%06ld %p",timer->deadline.tv_sec,timer->deadline.tv_usec,timer->ptr);
  co_list_delete(timers,node);
  
  timer->pending = false;
  return 1;

error:
  return 0;
}

int co_loop_set_timer	(	co_obj_t *	timer,
		long	msecs,
		co_obj_t *	context
	)

sets timer to expire in msecs from now

Parameters

timer	the timer to set
msecs	number of milliseconds
context	a co_obj_t context pointer (currently unused)

References co_loop_add_timer(), and co_loop_remove_timer().

                                                                          {
  co_timer_t *timer = (co_timer_t*)old_timer;
  struct timeval *deadline = &timer->deadline;
  
  if (timer->pending)
    co_loop_remove_timer((co_obj_t*)timer,context);
  
  _co_loop_gettime(&timer->deadline);
  
  deadline->tv_sec += msecs / 1000;
  deadline->tv_usec += (msecs % 1000) * 1000;
  
  if (deadline->tv_usec > 1000000) {
    deadline->tv_sec++;
    deadline->tv_usec %= 1000000;
  }
  
  return co_loop_add_timer((co_obj_t*)timer,context);
}

co_obj_t* co_timer_create	(	struct timeval	deadline,
		co_cb_t	timer_cb,
		void *	ptr
	)

malloc and initialize a timer

Parameters

size	timer struct size
proto	timer protocol

Referenced by _schedule().

                                                                                {
  co_timer_t *new_timer = h_calloc(1,sizeof(co_timer_t));
  
  new_timer->_header._type = _ext8;
  new_timer->_header._ref = 0;
  new_timer->_header._flags = 0;
  new_timer->_exttype = _co_timer;
  new_timer->_len = sizeof(co_timer_t);
  new_timer->pending = false;
  if (deadline.tv_sec > 0 || deadline.tv_usec > 0) {
    new_timer->deadline.tv_sec = deadline.tv_sec;
    new_timer->deadline.tv_usec = deadline.tv_usec;
  } else
    new_timer->deadline = (struct timeval){0};
  new_timer->timer_cb = timer_cb ? timer_cb : NULL;
  new_timer->ptr = ptr ? ptr : (void*)new_timer;
  
  return (co_obj_t*)new_timer;
}