* Allow to use GTK's main loop instead of the select stuff in
wait.c
-
-* need to close a lot of handles in w32-io.c
# AGE, set REVISION to 0.
# 3. Interfaces removed (BAD, breaks upward compatibility): Increment
# CURRENT, set AGE and REVISION to 0.
-AM_INIT_AUTOMAKE(gpgme,0.1.4a)
+AM_INIT_AUTOMAKE(gpgme,0.1.4b)
LIBGPGME_LT_CURRENT=2
LIBGPGME_LT_AGE=2
LIBGPGME_LT_REVISION=0
+2001-02-20 Werner Koch <wk@gnupg.org>
+
+ * w32-io.c (destroy_reader,kill_reader): New.
+ (create_reader, reader): Add a new event to stop the thread.
+ (_gpgme_io_close): Kill the reader thread.
+
+ * posix-io.c (_gpgme_io_select): Handle frozen fds here.
+ * 32-io.c (_gpgme_io_select): Ditto. Removed a bunch of unused code.
+
+ * wait.c: Reworked the whole thing.
+ * rungpg.c (_gpgme_gpg_new): Init pid to -1.
+ (_gpgme_gpg_release): Remove the process from the wait queue.
+
2001-02-19 Werner Koch <wk@gnupg.org>
+ * w32-io.c (_gpgme_io_set_close_notify): New.
+ (_gpgme_io_close): Do the notification.
+
* posix-io.c (_gpgme_io_select): Use a 1 sec timeout and not 200
microseconds.
* let autoconf (using the AM_PATH_GPGME macro) check that this
* header matches the installed library.
* Warning: Do not edit the next line. configure will do that for you! */
-#define GPGME_VERSION "0.1.4a"
+#define GPGME_VERSION "0.1.4b"
for ( i=0; i < nfds; i++ ) {
if ( fds[i].fd == -1 )
continue;
- if ( fds[i].for_read ) {
+ if ( fds[i].frozen ) {
+ DEBUG_ADD1 (dbg_help, "f%d ", fds[i].fd );
+ }
+ else if ( fds[i].for_read ) {
assert ( !FD_ISSET ( fds[i].fd, &readfds ) );
FD_SET ( fds[i].fd, &readfds );
if ( fds[i].fd > max_fd )
gpg->colon.fd[1] = -1;
gpg->cmd.fd = -1;
+ gpg->pid = -1;
+
/* allocate the read buffer for the status pipe */
gpg->status.bufsize = 1024;
gpg->status.readpos = 0;
free_argv (gpg->argv);
xfree (gpg->cmd.keyword);
+ if (gpg->pid != -1)
+ _gpgme_remove_proc_from_wait_queue ( gpg->pid );
if (gpg->status.fd[0] != -1 )
_gpgme_io_close (gpg->status.fd[0]);
if (gpg->status.fd[1] != -1 )
#define READBUF_SIZE 4096
+static struct {
+ int inuse;
+ int fd;
+ void (*handler)(int,void*);
+ void *value;
+} notify_table[256];
+DEFINE_STATIC_LOCK (notify_table_lock);
+
+
struct reader_context_s {
HANDLE file_hd;
HANDLE thread_hd;
DECLARE_LOCK (mutex);
-
+
+ int stop_me;
int eof;
int eof_shortcut;
int error;
HANDLE have_data_ev; /* manually reset */
HANDLE have_space_ev; /* auto reset */
+ HANDLE stopped;
size_t readpos, writepos;
char buffer[READBUF_SIZE];
};
DEBUG2 ("reader thread %p for file %p started", c->thread_hd, c->file_hd );
for (;;) {
LOCK (c->mutex);
- /* leave a one byte gap so that we can see wheter it is empty or full*/
+ /* leave a 1 byte gap so that we can see whether it is empty or full*/
if ((c->writepos + 1) % READBUF_SIZE == c->readpos) {
/* wait for space */
if (!ResetEvent (c->have_space_ev) )
DEBUG1 ("reader thread %p: got space", c->thread_hd );
LOCK (c->mutex);
}
+ if ( c->stop_me ) {
+ UNLOCK (c->mutex);
+ break;
+ }
nbytes = (c->readpos + READBUF_SIZE - c->writepos-1) % READBUF_SIZE;
if ( nbytes > READBUF_SIZE - c->writepos )
nbytes = READBUF_SIZE - c->writepos;
DEBUG2 ("reader thread %p: got %d bytes", c->thread_hd, (int)nread );
LOCK (c->mutex);
+ if (c->stop_me) {
+ UNLOCK (c->mutex);
+ break;
+ }
c->writepos = (c->writepos + nread) % READBUF_SIZE;
if ( !SetEvent (c->have_data_ev) )
DEBUG1 ("SetEvent failed: ec=%d", (int)GetLastError ());
if ( !SetEvent (c->have_data_ev) )
DEBUG1 ("SetEvent failed: ec=%d", (int)GetLastError ());
DEBUG1 ("reader thread %p ended", c->thread_hd );
+ SetEvent (c->stopped);
return 0;
}
c->file_hd = fd;
c->have_data_ev = CreateEvent (&sec_attr, TRUE, FALSE, NULL);
c->have_space_ev = CreateEvent (&sec_attr, FALSE, TRUE, NULL);
- if (!c->have_data_ev || !c->have_space_ev) {
+ c->stopped = CreateEvent (&sec_attr, TRUE, FALSE, NULL);
+ if (!c->have_data_ev || !c->have_space_ev || !c->stopped ) {
DEBUG1 ("** CreateEvent failed: ec=%d\n", (int)GetLastError ());
if (c->have_data_ev)
CloseHandle (c->have_data_ev);
if (c->have_space_ev)
CloseHandle (c->have_space_ev);
+ if (c->stopped)
+ CloseHandle (c->stopped);
xfree (c);
return NULL;
}
CloseHandle (c->have_data_ev);
if (c->have_space_ev)
CloseHandle (c->have_space_ev);
+ if (c->stopped)
+ CloseHandle (c->stopped);
xfree (c);
return NULL;
}
return c;
}
+static void
+destroy_reader (struct reader_context_s *c)
+{
+ if (c->have_space_ev)
+ SetEvent (c->have_space_ev);
+
+ DEBUG1 ("waiting for thread %p termination ...", c->thread_hd );
+ WaitForSingleObject (c->stopped, INFINITE);
+ DEBUG1 ("thread %p has terminated", c->thread_hd );
+
+ if (c->stopped)
+ CloseHandle (c->stopped);
+ if (c->have_data_ev)
+ CloseHandle (c->have_data_ev);
+ if (c->have_space_ev)
+ CloseHandle (c->have_space_ev);
+ CloseHandle (c->thread_hd);
+ DESTROY_LOCK (c->mutex);
+ xfree (c);
+}
+
/*
* Find a reader context or create a new one
}
+static void
+kill_reader (int fd)
+{
+ int i;
+
+ LOCK (reader_table_lock);
+ for (i=0; i < reader_table_size; i++ ) {
+ if (reader_table[i].used && reader_table[i].fd == fd ) {
+ destroy_reader (reader_table[i].context);
+ reader_table[i].context = NULL;
+ reader_table[i].used = 0;
+ break;
+ }
+ }
+ UNLOCK (reader_table_lock);
+}
+
+
int
_gpgme_io_read ( int fd, void *buffer, size_t count )
int
_gpgme_io_close ( int fd )
{
+ int i;
+ void (*handler)(int, void*) = NULL;
+ void *value = NULL;
+
if ( fd == -1 )
return -1;
DEBUG1 ("** closing handle for fd %d\n", fd);
- /* fixme: destroy thread */
+ kill_reader (fd);
+ LOCK (notify_table_lock);
+ for ( i=0; i < DIM (notify_table); i++ ) {
+ if (notify_table[i].inuse && notify_table[i].fd == fd) {
+ handler = notify_table[i].handler;
+ value = notify_table[i].value;
+ notify_table[i].handler = NULL;
+ notify_table[i].value = NULL;
+ notify_table[i].inuse = 0;
+ break;
+ }
+ }
+ UNLOCK (notify_table_lock);
+ if (handler)
+ handler (fd, value);
if ( !CloseHandle (fd_to_handle (fd)) ) {
DEBUG2 ("CloseHandle for fd %d failed: ec=%d\n",
return 0;
}
+int
+_gpgme_io_set_close_notify (int fd, void (*handler)(int, void*), void *value)
+{
+ int i;
+
+ assert (fd != -1);
+
+ LOCK (notify_table_lock);
+ for (i=0; i < DIM (notify_table); i++ ) {
+ if ( notify_table[i].inuse && notify_table[i].fd == fd )
+ break;
+ }
+ if ( i == DIM (notify_table) ) {
+ for (i=0; i < DIM (notify_table); i++ ) {
+ if ( !notify_table[i].inuse )
+ break;
+ }
+ }
+ if ( i == DIM (notify_table) ) {
+ UNLOCK (notify_table_lock);
+ return -1;
+ }
+ notify_table[i].fd = fd;
+ notify_table[i].handler = handler;
+ notify_table[i].value = value;
+ notify_table[i].inuse = 1;
+ UNLOCK (notify_table_lock);
+ DEBUG2 ("set notification for fd %d (idx=%d)", fd, i );
+ return 0;
+}
+
int
_gpgme_io_set_nonblocking ( int fd )
_gpgme_io_waitpid ( int pid, int hang, int *r_status, int *r_signal )
{
HANDLE proc = fd_to_handle (pid);
- int code, exc, ret = 0;
+ int code, ret = 0;
+ DWORD exc;
*r_status = 0;
*r_signal = 0;
int
_gpgme_io_select ( struct io_select_fd_s *fds, size_t nfds )
{
-#if 1
HANDLE waitbuf[MAXIMUM_WAIT_OBJECTS];
int waitidx[MAXIMUM_WAIT_OBJECTS];
int code, nwait;
DEBUG_BEGIN (dbg_help, "select on [ ");
any = any_write = 0;
nwait = 0;
+ count = 0;
for ( i=0; i < nfds; i++ ) {
if ( fds[i].fd == -1 )
continue;
+ fds[i].signaled = 0;
if ( fds[i].for_read || fds[i].for_write ) {
- if ( fds[i].for_read ) {
+ if ( fds[i].frozen ) {
+ DEBUG_ADD1 (dbg_help, "f%d ", fds[i].fd );
+ }
+ else if ( fds[i].for_read ) {
struct reader_context_s *c = find_reader (fds[i].fd,1);
if (!c) {
waitidx[nwait] = i;
waitbuf[nwait++] = c->have_data_ev;
}
+ DEBUG_ADD1 (dbg_help, "r%d ", fds[i].fd );
+ any = 1;
+ }
+ else if ( fds[i].for_write ) {
+ DEBUG_ADD1 (dbg_help, "w%d ", fds[i].fd );
+ any = 1;
+ /* no way to see whether a handle is ready for writing,
+ * so we signal them all */
+ fds[i].signaled = 1;
+ any_write =1;
+ count++;
}
- DEBUG_ADD2 (dbg_help, "%c%d ",
- fds[i].for_read? 'r':'w',fds[i].fd );
- any = 1;
}
- fds[i].signaled = 0;
}
DEBUG_END (dbg_help, "]");
if (!any)
return 0;
- count = 0;
- /* no way to see whether a handle is ready for writing, signal all */
- for ( i=0; i < nfds; i++ ) {
- if ( fds[i].fd == -1 )
- continue;
- if ( fds[i].for_write ) {
- fds[i].signaled = 1;
- any_write =1;
- count++;
- }
- }
code = WaitForMultipleObjects ( nwait, waitbuf, 0, any_write? 200:1000);
if ( code >= WAIT_OBJECT_0 && code < WAIT_OBJECT_0 + nwait ) {
/* This WFMO is a really silly function: It does return either
}
return count;
-#else /* This is the code we use */
- int i, any, count;
- int once_more = 0;
-
- DEBUG_SELECT ((stderr, "gpgme:fakedselect on [ "));
- any = 0;
- for ( i=0; i < nfds; i++ ) {
- if ( fds[i].fd == -1 )
- continue;
- if ( fds[i].for_read || fds[i].for_write ) {
- DEBUG_SELECT ((stderr, "%c%d ",
- fds[i].for_read? 'r':'w',fds[i].fd ));
- any = 1;
- }
- fds[i].signaled = 0;
- }
- DEBUG_SELECT ((stderr, "]\n" ));
- if (!any)
- return 0;
-
- restart:
- count = 0;
- /* no way to see whether a handle is ready fro writing, signal all */
- for ( i=0; i < nfds; i++ ) {
- if ( fds[i].fd == -1 )
- continue;
- if ( fds[i].for_write ) {
- fds[i].signaled = 1;
- count++;
- }
- }
-
- /* now peek on all read handles */
- for ( i=0; i < nfds; i++ ) {
- if ( fds[i].fd == -1 )
- continue;
- if ( fds[i].for_read ) {
- int navail;
-
- if ( !PeekNamedPipe (fd_to_handle (fds[i].fd),
- NULL, 0, NULL, &navail, NULL) ) {
- DEBUG1 ("select: PeekFile failed: ec=%d\n",
- (int)GetLastError ());
- }
- else if ( navail ) {
- DEBUG2 ("fd %d has %d bytes to read\n", fds[i].fd, navail );
- fds[i].signaled = 1;
- count++;
- }
- }
- }
- if ( !once_more && !count ) {
- /* once more but after relinquishing our timeslot */
- once_more = 1;
- Sleep (0);
- goto restart;
- }
-
- if ( count ) {
- DEBUG_SELECT ((stderr, "gpgme: signaled [ "));
- for ( i=0; i < nfds; i++ ) {
- if ( fds[i].fd == -1 )
- continue;
- if ( (fds[i].for_read || fds[i].for_write) && fds[i].signaled ) {
- DEBUG_SELECT ((stderr, "%c%d ",
- fds[i].for_read? 'r':'w',fds[i].fd ));
- }
- }
- DEBUG_SELECT ((stderr, "]\n" ));
- }
-
- return count;
-#endif
}
#endif /*HAVE_DOSISH_SYSTEM*/
#include "context.h"
#include "ops.h"
#include "wait.h"
+#include "sema.h"
#include "io.h"
-/* Fixme: implement the following stuff to make the code MT safe.
- * To avoid the need to link against a specific threads lib, such
- * an implementation should require the caller to register a function
- * which does this task.
- * enter_crit() and leave_crit() are used to embrace an area of code
- * which should be executed only by one thread at a time.
- * lock_xxxx() and unlock_xxxx() protect access to an data object.
- * */
-#define enter_crit() do { } while (0)
-#define leave_crit() do { } while (0)
-#define lock_table() do { } while (0)
-#define unlock_table() do { } while (0)
+struct wait_item_s;
+struct proc_s;
+static struct proc_s *proc_queue;
+DEFINE_STATIC_LOCK (proc_queue_lock);
+
+static int fd_table_size;
+static struct io_select_fd_s *fd_table;
+DEFINE_STATIC_LOCK (fd_table_lock);
+
+static void (*idle_function) (void);
+
+
+struct proc_s {
+ struct proc_s *next;
+ int pid;
+ GpgmeCtx ctx;
+ struct wait_item_s *handler_list;
+ int ready;
+};
struct wait_item_s {
- volatile int active;
+ struct wait_item_s *next;
int (*handler)(void*,int,int);
void *handler_value;
- int pid;
int inbound; /* this is an inbound data handler fd */
- GpgmeCtx ctx;
+ struct proc_s *proc; /* backlink */
+ int ready;
+ int frozen; /* copy of the frozen flag from the fd_table */
};
-static int fd_table_size;
-static struct io_select_fd_s *fd_table;
-static void (*idle_function) (void);
static int do_select ( void );
static void run_idle (void);
-static struct wait_item_s *
-queue_item_from_context ( GpgmeCtx ctx )
-{
- struct wait_item_s *q;
- int i;
-
- for (i=0; i < fd_table_size; i++ ) {
- if ( fd_table[i].fd != -1 && (q=fd_table[i].opaque) && q->ctx == ctx )
- return q;
- }
- return NULL;
-}
-
-
+/* only to be called with a locked proc_queue */
static int
-count_active_and_thawed_fds ( int pid )
+count_running_fds ( struct proc_s *proc )
{
struct wait_item_s *q;
- int i, count = 0;
-
- for (i=0; i < fd_table_size; i++ ) {
- if ( fd_table[i].fd != -1 && (q=fd_table[i].opaque)
- && q->active && !fd_table[i].frozen && q->pid == pid )
+ int count = 0;
+
+ for (q=proc->handler_list; q; q=q->next) {
+ if ( !q->frozen && !q->ready )
count++;
}
return count;
}
-/* remove the given process from the queue */
-/* FIXME: We should do this on demand from rungpg.c */
+/* only to be called with a locked proc_queue */
static void
-remove_process ( int pid )
+set_process_ready ( struct proc_s *proc )
{
- struct wait_item_s *q;
+ struct wait_item_s *q, *q2;
int i;
- for (i=0; i < fd_table_size; i++ ) {
- if (fd_table[i].fd != -1 && (q=fd_table[i].opaque) && q->pid == pid ) {
- xfree (q);
- fd_table[i].opaque = NULL;
- fd_table[i].fd = -1;
+ assert (proc);
+ DEBUG2 ("set_process_ready(%p) pid=%d", proc, proc->pid );
+ LOCK (fd_table_lock);
+ for (q = proc->handler_list; q; q=q2) {
+ q2 = q->next;
+ for (i=0; i < fd_table_size; i++ ) {
+ if (fd_table[i].fd != -1 && q == fd_table[i].opaque ) {
+ fd_table[i].opaque = NULL;
+ fd_table[i].fd = -1;
+ }
}
+ xfree (q);
}
+ UNLOCK (fd_table_lock);
+ proc->handler_list = NULL;
+ proc->ready = 1;
}
+void
+_gpgme_remove_proc_from_wait_queue ( int pid )
+{
+ struct proc_s *proc, *last;
+
+ DEBUG1 ("removing process %d", pid );
+ LOCK (proc_queue_lock);
+ for (last=NULL, proc=proc_queue; proc; last = proc, proc = proc->next ) {
+ if (proc->pid == pid ) {
+ set_process_ready (proc);
+ if (!last)
+ proc_queue = proc->next;
+ else
+ last->next = proc->next;
+ xfree (proc);
+ break;
+ }
+ }
+ UNLOCK (proc_queue_lock);
+}
/**
GpgmeCtx
_gpgme_wait_on_condition ( GpgmeCtx c, int hang, volatile int *cond )
{
- struct wait_item_s *q;
-
+ DEBUG3 ("waiting... ctx=%p hang=%d cond=%p", c, hang, cond );
do {
int did_work = do_select();
+ int any = 0;
+ struct proc_s *proc;
if ( cond && *cond )
hang = 0;
-
- if ( !did_work ) {
- /* We did no read/write - see whether the process is still
- * alive */
- assert (c); /* !c is not yet implemented */
- q = queue_item_from_context ( c );
- if (q) {
- if ( !count_active_and_thawed_fds (q->pid) ) {
- remove_process (q->pid);
- hang = 0;
+ else {
+ LOCK (proc_queue_lock);
+ for (proc=proc_queue; proc; proc = proc->next ) {
+ if ( !proc->ready && !count_running_fds (proc) ) {
+ set_process_ready (proc);
}
+ if (c && proc->ready && proc->ctx == c)
+ hang = 0;
+ if ( !proc->ready )
+ any = 1;
}
- else
+ UNLOCK (proc_queue_lock);
+ if (!any)
hang = 0;
}
+ /* fixme: we should check here for hanging processes */
+
if (hang)
run_idle ();
} while (hang && !c->cancel );
static int
do_select ( void )
{
- struct wait_item_s *q;
int i, n;
int any=0;
for (i=0; i < fd_table_size && n; i++ ) {
if ( fd_table[i].fd != -1 && fd_table[i].signaled
&& !fd_table[i].frozen ) {
- q = fd_table[i].opaque;
+ struct wait_item_s *q;
+
assert (n);
n--;
- if ( q->active )
- any = 1;
- if ( q->active && q->handler (q->handler_value,
- q->pid, fd_table[i].fd ) ) {
- DEBUG1 ("setting fd %d inactive", fd_table[i].fd );
- q->active = 0;
+
+ q = fd_table[i].opaque;
+ assert ( q );
+ assert ( q->proc );
+ assert ( !q->ready );
+ any = 1;
+ if ( q->handler (q->handler_value,
+ q->proc->pid, fd_table[i].fd ) ) {
+ DEBUG2 ("setting fd %d (q=%p) ready", fd_table[i].fd, q );
+ q->ready = 1;
+ /* free the table entry*/
+ LOCK (fd_table_lock);
fd_table[i].for_read = 0;
fd_table[i].for_write = 0;
fd_table[i].fd = -1;
+ fd_table[i].opaque = NULL;
+ UNLOCK (fd_table_lock);
}
}
}
{
GpgmeCtx ctx = opaque;
struct wait_item_s *q;
+ struct proc_s *proc;
int i;
assert (opaque);
assert (handler);
-
+
+ /* Allocate a structure to hold info about the handler */
q = xtrycalloc ( 1, sizeof *q );
if ( !q )
return mk_error (Out_Of_Core);
q->inbound = inbound;
q->handler = handler;
q->handler_value = handler_value;
- q->pid = pid;
- q->ctx = ctx;
- q->active = 1;
- lock_table ();
+ /* Put this into the process queue */
+ LOCK (proc_queue_lock);
+ for (proc=proc_queue; proc && proc->pid != pid; proc = proc->next)
+ ;
+ if (!proc) { /* a new process */
+ proc = xtrycalloc ( 1, sizeof *proc );
+ if (!proc) {
+ UNLOCK (proc_queue_lock);
+ return mk_error (Out_Of_Core);
+ }
+ proc->pid = pid;
+ proc->ctx = ctx;
+ proc->next = proc_queue;
+ proc_queue = proc;
+ }
+ assert (proc->ctx == ctx);
+ q->proc = proc;
+ q->next = proc->handler_list;
+ proc->handler_list = q;
+ UNLOCK (proc_queue_lock);
+
+ LOCK (fd_table_lock);
again:
for (i=0; i < fd_table_size; i++ ) {
if ( fd_table[i].fd == -1 ) {
fd_table[i].signaled = 0;
fd_table[i].frozen = 0;
fd_table[i].opaque = q;
- unlock_table ();
+ UNLOCK (fd_table_lock);
return 0;
}
}
}
}
- unlock_table ();
+ UNLOCK (fd_table_lock);
xfree (q);
+ /* FIXME: remove the proc table entry */
return mk_error (Too_Many_Procs);
}
{
int i;
- lock_table ();
+ LOCK (fd_table_lock);
for (i=0; i < fd_table_size; i++ ) {
if ( fd_table[i].fd == fd ) {
+ struct wait_item_s *q;
+
fd_table[i].frozen = 1;
- DEBUG1 ("fd %d frozen", fd );
+ if ( (q=fd_table[i].opaque) )
+ q->frozen = 1;
+ DEBUG2 ("fd %d frozen (q=%p)", fd, q );
break;
}
}
- unlock_table ();
+ UNLOCK (fd_table_lock);
}
void
{
int i;
- lock_table ();
+ LOCK (fd_table_lock);
for (i=0; i < fd_table_size; i++ ) {
if ( fd_table[i].fd == fd ) {
+ struct wait_item_s *q;
+
fd_table[i].frozen = 0;
- DEBUG1 ("fd %d thawed", fd );
+ if ( (q=fd_table[i].opaque) )
+ q->frozen = 0;
+ DEBUG2 ("fd %d thawed (q=%p)", fd, q );
break;
}
}
- unlock_table ();
+ UNLOCK (fd_table_lock);
}
#include "gpgme.h"
-#define SIZEOF_WAIT_QUEUE 10
+void _gpgme_remove_proc_from_wait_queue ( int pid );
-
-GpgmeError _gpgme_register_pipe_handler(
+GpgmeError _gpgme_register_pipe_handler (
void *opaque,
int (*handler)(void*,int,int),
void *handler_value,
int pid, int fd, int inbound );
-
-
#endif /* WAIT_H */
projects / gpgme.git / commitdiff