This is TMConnectionRunner.m in view mode; [Download] [Up]
/* Implementation of the TMConnectionRunner class.
Written by Tiggr <tiggr@es.ele.tue.nl> and Michael <michael@thi.nl>
Copyright (C) 1994, 1995 Pieter J. Schoenmakers and Michael L.H. Brouwer.
All rights reserved.
This file is part of RUNNER.
RUNNER is free software; you can redistribute it and/or modify it
under the terms of the GNU Library General Public License as
published by the Free Software Foundation; either version 2, or (at
your option) any later version.
RUNNER 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 Library General Public
License for more details.
You should have received a copy of the GNU Library General Public
License along with RUNNER; see the file COPYING.LIB. If not, write
to the Free Software Foundation, 59 Temple Place - Suite 330, Boston,
MA 02111-1307, USA.
$Id: TMConnectionRunner.m,v 1.2 1995/11/29 17:40:51 tiggr Exp $ */
#import <mach/mach.h>
#import <mach/notify.h>
#import <mach/mach_error.h>
#import <machkit/NXPort.h>
#import <remote/NXConnection.h>
#import <remote/NXProxy.h>
#import <sys/time.h>
#ifdef HAVE_FOUNDATION
#import <foundation/NSAutoreleasePool.h>
#endif
#import "TMConnectionRunner.h"
/* The structure used to record information on a port. */
struct dps_registered_port
{
/* The port proc to be invoked. */
DPSPortProc proc;
/* The user definable argument for the PROC. */
void *user_data;
/* The default timeout (in milliseconds) to start counting down from when
triggered. If 0, this entry is never timed out and not even present on
the delta list. */
int timeout_init;
/* The current timeout value, relative to the previous entry in the delta
list. */
int timeout_delta;
/* The delta dll used for timeouts. */
struct dps_registered_port *next, **prev;
};
/* How nice of NeXT not to define this struct. Now we can do it, while
`dpsclient/dpsNeXT.h' can still be included. */
struct __DPSTimedEntry
{
/* The port used to timeout in this entry. We could of course use a
single shared port and compute the timeout to use for the shared port
each time by ourself, but that would be duplicating the functionality
already present for port timeouts. Besides, ports are cheap. */
port_t port;
/* The proc to invoke. */
DPSTimedEntryProc proc;
/* The user definable argument for the PROC. */
void *user_data;
/* The time (in milliseconds) between successive invocations of this
proc. */
int period;
};
/* The shared instance of the TMConnectionRunner. */
static id shared_instance;
/* Forward declaration. */
static void notification_handler (msg_header_t *msg, void *userData);
void
DPSAddPort (port_t new_port, DPSPortProc handler, int max_size,
void *user_data, int priority)
{
[[TMConnectionRunner sharedRunner] addPort: new_port
handler: handler data: user_data timeout: 0];
} /* DPSAddPort */
void
DPSRemovePort(port_t port)
{
[[TMConnectionRunner sharedRunner] removePort: port];
} /* DPSRemovePort */
DPSTimedEntry
DPSAddTimedEntry (double period, DPSTimedEntryProc handler,
void *user_data, int priority)
{
return [[TMConnectionRunner sharedRunner] addTimedEntry: 1000 * period
handler: handler data: user_data];
} /* DPSAddTimedEntry */
void
DPSRemoveTimedEntry (DPSTimedEntry te)
{
[[TMConnectionRunner sharedRunner] removeTimedEntry: te];
} /* DPSRemoveTimedEntry */
void
DPSStartTimeout (port_t port, int timeout)
{
if (timeout == -1)
[[TMConnectionRunner sharedRunner] startTimeout: port];
else
[[TMConnectionRunner sharedRunner] startTimeout: port timeout: timeout];
} /* DPSStartTimeout */
void
DPSCancelTimeout (port_t port)
{
[[TMConnectionRunner sharedRunner] cancelTimeout: port];
} /* DPSCancelTimeout */
static void
timed_entry_handler (msg_header_t *msg, void *user_data)
{
DPSTimedEntry te = user_data;
struct timeval start, end;
double now;
time_t ms;
if (msg)
{
/* Someone's doing naughty things with the wrong port. */
abort ();
}
gettimeofday (&start, NULL);
now = start.tv_sec + start.tv_usec / 1e6;
te->proc (te, now, te->user_data);
gettimeofday (&end, NULL);
/* Initiate the next invocation. */
ms = (1000 * (end.tv_sec - start.tv_sec)
+ (end.tv_usec - start.tv_usec + 500) / 1000);
[shared_instance startTimeout: te->port
timeout: ((ms > te->period) ? 0 : te->period - ms)];
} /* timed_entry_handler */
@implementation TMConnectionRunner
/******************** private methods ********************/
/* */
-(struct dps_registered_port *) registrationOfPort: (port_t) port
{
struct dps_registered_port *reg = [ports valueForKey: (void *) port];
if (!reg)
LogError ("no port %d", port);
return (reg);
} /* -registrationOfPort: */
-(void) handleNotification: (notification_t *) nt
{
if (nt->notify_header.msg_id == NOTIFY_PORT_DELETED)
{
struct dps_registered_port *reg;
[(NXPort *) [NXPort newFromMachPort: nt->notify_port] invalidate];
reg = [ports valueForKey: (void *) nt->notify_port];
if (reg)
[self removePort: nt->notify_port];
}
} /* -handleNotification: */
/* Remove the REG from the timeout dll and return it. */
-(struct dps_registered_port *)
removeFromTimeoutQueue: (struct dps_registered_port *) reg
{
*reg->prev = reg->next;
if (reg->next)
{
reg->next->prev = reg->prev;
reg->next->timeout_delta += reg->timeout_delta;
}
else
timeout_tail = reg->prev;
reg->prev = NULL;
reg->next = NULL;
return (reg);
} /* -removeFromTimeoutQueue: */
-(void) startTimeout: (int) timeout
ofRegistration: (struct dps_registered_port *) reg
{
struct dps_registered_port **preg;
int cum;
if (reg->prev)
{
LogError ("%d", timeout);
[self removeFromTimeoutQueue: reg];
}
for (cum = 0, preg = &timeout_head;
*preg && cum + (*preg)->timeout_delta < timeout;
cum += (*preg)->timeout_delta, preg = &(*preg)->next);
reg->next = *preg;
reg->prev = preg;
*preg = reg;
reg->timeout_delta = timeout - cum;
if (reg->next)
{
reg->next->timeout_delta -= reg->timeout_delta;
reg->next->prev = ®->next;
}
else
timeout_tail = ®->next;
} /* -startTimeout:ofRegistration: */
/******************** public methods ********************/
/* Make sure only one runner is ever active. */
+sharedRunner
{
if (!shared_instance)
shared_instance = [[self alloc] init];
return (shared_instance);
} /* +sharedRunner */
-(void) addPort: (port_t) port handler: (DPSPortProc) handler
data: (void *) data timeout: (int) timeout
{
struct dps_registered_port *reg;
kern_return_t kr;
kr = port_set_add (task_self (), set, port);
if (kr != KERN_SUCCESS)
LogError ("port=%d set=%d: %s", port, set, mach_error_string (kr));
else
{
reg = malloc (sizeof (*reg));
reg->proc = handler;
reg->user_data = data;
reg->timeout_init = timeout;
reg->timeout_delta = 0;
reg->next = NULL;
reg->prev = NULL;
[(HashTable *) ports insertKey: (void *) port value: reg];
}
} /* -addPort:handler:data:timeout: */
-(DPSTimedEntry) addTimedEntry: (int) period
handler: (DPSTimedEntryProc) handler
data: (void *) user_data
{
DPSTimedEntry te = malloc (sizeof (*te));
kern_return_t kr = port_allocate (task_self (), &te->port);
if (kr != KERN_SUCCESS)
{
LogError ("port_allocate: %s (%d)", mach_error_string (kr), kr);
free (te);
return (NULL);
}
te->proc = handler;
te->user_data = user_data;
te->period = period;
[self addPort: te->port handler: timed_entry_handler data: te timeout: 0];
[self startTimeout: te->port timeout: te->period];
return (te);
} /* -addTimedEntry:handler:data: */
-(void) cancelTimeout: (port_t) port
{
struct dps_registered_port *reg = [self registrationOfPort: port];
if (reg)
if (!reg->prev)
LogError ("port %d not in timeout queue", port);
else
[self removeFromTimeoutQueue: reg];
} /* -cancelTimeout: */
-init
{
kern_return_t kr;
port_t p;
if (![super init])
return (nil);
timeout_tail = &timeout_head;
p = task_notify ();
if (p)
kr = KERN_SUCCESS;
else
{
kr = port_allocate (task_self (), &p);
if (kr == KERN_SUCCESS)
kr = task_set_special_port (task_self (), TASK_NOTIFY_PORT, p);
}
if (kr == KERN_SUCCESS)
{
kr = port_set_allocate (task_self (), &set);
if (kr == KERN_SUCCESS)
port_set_add (task_self (), set, p);
}
if (kr != KERN_SUCCESS)
[self error: "port manipulation: %s", mach_error_string (kr)];
ports = [[HashTable alloc] initKeyDesc: "i" valueDesc: "!"];
[self addPort: task_notify () handler: notification_handler
data: self timeout: 0];
return (self);
} /* -init */
-(void) removePort: (port_t) port
{
struct dps_registered_port *reg;
port_set_remove (task_self (), port);
reg = [self registrationOfPort: port];
if (reg->prev)
[self removeFromTimeoutQueue: reg];
[ports removeKey: (void *) port];
free (reg);
} /* -removePort: */
-(void) removeTimedEntry: (DPSTimedEntry) te
{
kern_return_t kr;
[self removePort: te->port];
kr = port_deallocate (task_self (), te->port);
if (kr != KERN_SUCCESS)
LogError ("port_deallocate (%d): %s (%d)",
te->port, mach_error_string (kr), kr);
free (te);
} /* -removeTimedEntry: */
-(void) run
{
struct dps_registered_port *msg_reg;
msg_header_t *msg = alloca (MSG_SIZE_MAX);
struct timeval previous, now;
int used_timeout;
msg_return_t mr;
#ifdef HAVE_FOUNDATION
NSAutoreleasePool *pool;
#endif
gettimeofday (&previous, NULL);
for (;;)
{
#ifdef HAVE_FOUNDATION
pool = [[NSAutoreleasePool alloc] init];
#endif
msg->msg_local_port = set;
msg->msg_size = MSG_SIZE_MAX;
/* If we haven't got things to do, wait indefinitely. */
if (!timeout_head)
{
used_timeout = 0;
mr = msg_receive (msg, 0, 0);
}
else
{
used_timeout = timeout_head->timeout_delta;
mr = msg_receive (msg, RCV_TIMEOUT,
(timeout_head->timeout_delta < 0
? 0 : timeout_head->timeout_delta));
}
msg_reg = NULL;
if (mr == RCV_SUCCESS)
{
msg_reg = [self registrationOfPort: msg->msg_local_port];
/* Remove the timeout, if neccessary. */
if (msg_reg && msg_reg->prev)
[self removeFromTimeoutQueue: msg_reg];
}
else if (mr != RCV_TIMED_OUT)
LogError ("msg_receive: %s", mach_error_string (mr));
/* Handle the timeout queue. */
gettimeofday (&now, NULL);
if (timeout_head)
{
timeout_head->timeout_delta
-= ((now.tv_usec - previous.tv_usec) / 1000
+ (now.tv_sec - previous.tv_sec) * 1000);
if (timeout_head->timeout_delta <= 0)
{
struct dps_registered_port *reg;
int i;
/* Run at most the number of zero-delta entries present at
_this_ moment. */
for (i = 0, reg = timeout_head; reg && reg->timeout_delta <= 0;
i++, reg = reg->next);
/* Run elapsed timeouts. */
while (timeout_head && timeout_head->timeout_delta <= 0 && i--)
{
reg = [self removeFromTimeoutQueue: timeout_head];
reg->proc (NULL, reg->user_data);
}
}
}
previous = now;
/* Invoke the handler. */
if (msg_reg)
msg_reg->proc (msg, msg_reg->user_data);
#ifdef HAVE_FOUNDATION
[pool release];
#endif
}
} /* -run */
-(void) startTimeout: (port_t) port
{
struct dps_registered_port *reg = [self registrationOfPort: port];
if (reg)
[self startTimeout: reg->timeout_init ofRegistration: reg];
else
LogError ("port=%d not registered", port);
} /* -startTimeout: */
-(void) startTimeout: (port_t) port timeout: (int) timeout
{
struct dps_registered_port *reg = [self registrationOfPort: port];
if (reg)
[self startTimeout: timeout ofRegistration: reg];
else
LogError ("port=%d timeout=%d not registered", port, timeout);
} /* -startTimeout: */
@end
static void
notification_handler (msg_header_t *msg, void *userData)
{
[(id) userData handleNotification: (void *) msg];
} /* notification_handler */
These are the contents of the former NiCE NeXT User Group NeXTSTEP/OpenStep software archive, currently hosted by Netfuture.ch.