gdk_system.mx

这个是内存数据库中的一个管理工具

@' The contents of this file are subject to the MonetDB Public License
@' Version 1.1 (the "License"); you may not use this file except in
@' compliance with the License. You may obtain a copy of the License at
@' http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
@'
@' Software distributed under the License is distributed on an "AS IS"
@' basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
@' License for the specific language governing rights and limitations
@' under the License.
@'
@' The Original Code is the MonetDB Database System.
@'
@' The Initial Developer of the Original Code is CWI.
@' Portions created by CWI are Copyright (C) 1997-2007 CWI.
@' All Rights Reserved.

@f gdk_system
@a Niels Nes, Peter Boncz
@+ Threads
This file contains a wrapper layer for threading, hence the underscore 
convention MT_x (Multi-Threading).  As all platforms that MonetDB runs 
on now support POSIX Threads (pthreads), this wrapping layer has become
rather thin.

In the late 1990s when multi-threading support was introduced in MonetDB, 
pthreads was just emerging as a standard API and not widely adaopted yet. 
The earliest MT implementation focused on SGI Unix and provided multi-
threading using multiple processses, and shared memory. 

One of the relics of this model, namely the need to pre-allocate locks and 
semaphores, and consequently a maximum number of them, has been removed in 
the latest iteration of this layer.

@{
@h
#ifndef _GDK_SYSTEM_H_
#define _GDK_SYSTEM_H_

#ifdef NATIVE_WIN32
#ifndef LIBGDK
#define gdk_export extern __declspec(dllimport)
#else
#define gdk_export extern __declspec(dllexport)
#endif
#else
#define gdk_export extern
#endif

#ifdef HAVE_UNISTD_H
# include <unistd.h>		/* io functions */
#endif
#ifdef HAVE_IO_H
# include <io.h>
#endif
#ifdef HAVE_FCNTL_H
#include <fcntl.h>
#endif

@- pthreads Includes and Definitions
@h
#include <sys/types.h>
#ifdef HAVE_SIGNAL_H
# include <signal.h>
#endif

#ifdef HAVE_PTHREAD_H
/* don't re-include config.h; on Windows, don't redefine pid_t in an
   incompatible way 
 */
#undef HAVE_CONFIG_H
#ifdef pid_t
#undef pid_t
#endif
#include <sched.h>
#include <pthread.h>
#ifndef WIN32
/* Linux gprof messes up on multithreaded programs */
#ifdef PROFILE
#define pthread_create gprof_pthread_create
#endif
#endif
#endif

#ifdef HAVE_SEMAPHORE_H
# include <semaphore.h>
#endif

/* new pthread interface, where the thread id changed to a struct */
#ifdef PTW32_VERSION
#define PTW32 1
#endif

#if !(defined(HAVE_SYSCONF) && defined(_SC_PHYS_PAGES)) && defined(HAVE_GETRLIMIT) && defined(HAVE_SYS_RESOURCE_H)
# include <sys/resource.h>
#endif

/* debug and errno integers */
gdk_export int GDKdebug;
#define MT_geterrno()	errno
#define MT_seterrno(x)	errno=x

#define MT_log(_impl, _object, _action, _caller, _fp) do { if (GDKdebug & 1024) { fprintf(_fp, "%s: " _action "(" PTRFMT ")\n", _caller, PTRFMTCAST(void*) _object); fflush(_fp); } _impl; } while (0)

/* virtual memory defines */
gdk_export size_t _MT_npages;
gdk_export size_t _MT_pagesize;
#define MT_pagesize()	_MT_pagesize
#define MT_npages()	_MT_npages
@h
/* API */
gdk_export void    MT_init(void);     /*  init the package. */
gdk_export int     MT_alive(int pid); /* OS independent way to check if some process is still alive. */

#define MT_initialized() (MT_pagesize() > 0)

@- MT Thread Api
@h
typedef size_t MT_Id; /* thread number. will not be zero*/

gdk_export int     MT_create_thread(MT_Id *t, void (*function) (void *), void *arg);
gdk_export void    MT_exit_thread(int status);
gdk_export void    MT_global_exit(int status);
gdk_export MT_Id   MT_getpid(void);
gdk_export int     MT_kill_thread(MT_Id t);
#ifdef HAVE_PTHREAD_SIGMASK
gdk_export void    MT_thread_sigmask(sigset_t * new_mask, sigset_t * orig_mask);
#endif

#if SIZEOF_VOID_P == 4
/* "limited" stack size on 32-bit systems */
/* to avoid address space fragmentation   */
#define THREAD_STACK_SIZE	((size_t)2*512*1024)
#else
/* "increased" stack size on 64-bit systems    */
/* since some compilers seem to require this   */
/* for burg-generated code in pathfinder       */
/* and address space fragmentation is no issue */
#define THREAD_STACK_SIZE	((size_t)4*512*1024)
#endif


@- MT Lock API
@h
typedef pthread_mutex_t MT_Lock;

#define MT_lock_init(l)      pthread_mutex_init((pthread_mutex_t*) l, 0)
#define MT_lock_destroy(l)   pthread_mutex_destroy((pthread_mutex_t*) l)
#define MT_lock_set(l,n)     MT_log(pthread_mutex_lock((pthread_mutex_t *) l), l, "MT_set_lock", n, stderr)
#define MT_lock_unset(l,n)   MT_log(pthread_mutex_unlock((pthread_mutex_t *) l), l, "MT_unset_lock", n, stderr)
#define MT_lock_try(l)       pthread_mutex_trylock((pthread_mutex_t *) l)
#define MT_lock_dump(l,fp,n) MT_log(/*nothing*/, &l, "MT_dump_lock", n, fp)

/* backward compatability API */
#define MT_init_lock(l)      MT_lock_init(&l)   
#define MT_destroy_lock(l)   MT_lock_destroy(&l)   
#define MT_set_lock(l,n)     MT_lock_set(&l,n)     
#define MT_unset_lock(l,n)   MT_lock_unset(&l,n)   
#define MT_try_lock(l)       MT_lock_try(&l)       
#define MT_dump_lock(l,fp,n) MT_lock_dump(&l,fp,n) 

gdk_export MT_Lock MT_system_lock;

@- MT Semaphore API
@h
#if defined(_AIX) || defined(__APPLE_CC__)
typedef struct {
	int cnt;
	pthread_mutex_t mutex;
	pthread_cond_t cond;
} pthread_sema_t;

gdk_export void    pthread_sema_init(pthread_sema_t *s, int flag, int nresources);
gdk_export void    pthread_sema_destroy(pthread_sema_t *s);
gdk_export void    pthread_sema_up(pthread_sema_t *s);
gdk_export void    pthread_sema_down(pthread_sema_t *s);
#else
#define pthread_sema_t       sem_t
#define pthread_sema_init    sem_init
#define pthread_sema_destroy sem_destroy
#define pthread_sema_up      sem_post
#define pthread_sema_down(x) while(sem_wait(x)) 
#endif

typedef pthread_sema_t MT_Sema;

#define MT_sema_init(s,nr)   pthread_sema_init(s,0,nr)
#define MT_sema_destroy(s)   pthread_sema_destroy(s)
#define MT_sema_up(s,n)      MT_log(pthread_sema_up(s), s, "MT_up_sema", n, stderr)
#define MT_sema_down(s,n)    MT_log(pthread_sema_down(s), s, "MT_down_sema", n, stderr)
#define MT_sema_dump(s,fp,n) MT_log(/*nothing*/, s, "MT_dump_sema", n, fp)

/* backward compatability API */
#define MT_init_sema(s,nr)   MT_sema_init(&s,nr)
#define MT_destroy_sema(s)   MT_sema_destroy(&s)
#define MT_up_sema(s,n)      MT_sema_up(&s,n)
#define MT_down_sema(s,n)    MT_sema_down(&s,n)
#define MT_dump_sema(s,fp,n) MT_sema_dump(&s,fp,n)

@- MT Conditional Variable API
@h
typedef pthread_cond_t MT_Cond;

#define MT_cond_init(c)      pthread_cond_init((pthread_cond_t*) c, NULL)
#define MT_cond_destroy(c)   pthread_cond_destroy((pthread_cond_t*) c)
#define MT_cond_signal(c,n)  MT_log(pthread_cond_signal((pthread_cond_t*) c), c, "MT_signal_cond", n, stderr)
#define MT_cond_wait(c,l,n)  MT_log(pthread_cond_wait((pthread_cond_t*) c, (pthread_mutex_t *) l), c, "MT_wait_cond", n, stderr)

/* backward compatability API */
#define MT_init_cond(c)      MT_cond_init(&c)
#define MT_destroy_cond(c)   MT_cond_destroy(&c)   
#define MT_signal_cond(c,n)  MT_cond_signal(&c,n)  
#define MT_wait_cond(c,l,n)  MT_cond_wait(&c,&l,n)  

#endif /*_GDK_SYSTEM_H_*/


@- Mthreads Routine implementations
@c
#include "monetdb_config.h"
#include "gdk.h"

size_t _MT_pagesize = 0;	/* variable holding memory size */
size_t _MT_npages = 0;		/* variable holding page size */

MT_Lock MT_system_lock = PTHREAD_MUTEX_INITIALIZER;

void
MT_init(void)
{
#ifdef HAVE_GETSYSTEMINFO
	{
		SYSTEM_INFO sysInfo;

		GetSystemInfo(&sysInfo);
		_MT_pagesize = sysInfo.dwPageSize;
	}
#else
#if defined(HAVE_SYSCONF) && defined(_SC_PAGESIZE)
	_MT_pagesize = sysconf(_SC_PAGESIZE);
#endif
#endif
	if (_MT_pagesize <= 0)
		_MT_pagesize = 4096;	/* default */

#ifdef HAVE_GLOBALMEMORYSTATUSEX
	{
		MEMORYSTATUSEX memStatEx;

		memStatEx.dwLength = sizeof(memStatEx);
		if (GlobalMemoryStatusEx(&memStatEx))
			_MT_npages = (size_t) (memStatEx.ullTotalPhys / _MT_pagesize);
	}
#endif
#ifdef HAVE_GLOBALMEMORYSTATUS
	if (_MT_npages <= 0) {
		MEMORYSTATUS memStat;

		GlobalMemoryStatus(&memStat);
		_MT_npages = memStat.dwTotalPhys / _MT_pagesize;
	}
#endif
#if defined(HAVE_SYSCONF) && defined(_SC_PHYS_PAGES)
	_MT_npages = sysconf(_SC_PHYS_PAGES);
#else
#ifdef HAVE_GETRLIMIT
	{
		struct rlimit rl;

		getrlimit(RLIMIT_RSS, &rl);
		_MT_npages = rl.rlim_cur / _MT_pagesize;
	}
#endif
#endif
}

#ifdef HAVE_PTHREAD_SIGMASK
void
MT_thread_sigmask(sigset_t * new_mask, sigset_t * orig_mask)
{
	(void) sigdelset(new_mask, SIGQUIT);
	(void) sigdelset(new_mask, SIGALRM);	/* else sleep doesn't work */
	(void) pthread_sigmask(SIG_SETMASK, new_mask, orig_mask);
}
#endif

int
MT_create_thread(MT_Id *t, void (*f) (void *), void *arg)
{
#ifdef HAVE_PTHREAD_SIGMASK
	sigset_t new_mask, orig_mask;
#endif
	pthread_attr_t attr;
	pthread_t newt;
	int ret;

#ifdef HAVE_PTHREAD_SIGMASK
	(void) sigfillset(&new_mask);
	MT_thread_sigmask(&new_mask, &orig_mask);
#endif
	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	pthread_attr_setstacksize(&attr, 4 * THREAD_STACK_SIZE);
	ret = pthread_create(&newt, &attr, (void *(*)(void *)) f, arg);
	if (ret == 0)
#ifdef PTW32
		*t = (MT_Id) (((size_t) newt.p) + 1);	/* use pthread-id + 1 */
#else
		*t = (MT_Id) (((size_t) newt) + 1);	/* use pthread-id + 1 */
#endif
#ifdef HAVE_PTHREAD_SIGMASK
	MT_thread_sigmask(&orig_mask, NULL);
#endif
	return ret;
}

void
MT_global_exit(int s)
{
	exit(s);
}


void
MT_exit_thread(int s)
{
	int st = s;
	pthread_exit(&st);
}


int
MT_kill_thread(MT_Id t)
{
#ifdef HAVE_PTHREAD_KILL
	thread_t id = (thread_t) (t - 1);

	return pthread_kill((pthread_t) id, SIGHUP);
#else
	(void) t;
	return -1;		/* XXX */
#endif
}
#if defined(_AIX) || defined(__APPLE_CC__)
void
pthread_sema_init(pthread_sema_t *s, int flag, int nresources)
{
	(void) flag;
	s->cnt = nresources;
	pthread_mutex_init(&(s->mutex), 0);
	pthread_cond_init(&(s->cond), 0);
}

void
pthread_sema_destroy(pthread_sema_t *s)
{
	pthread_mutex_destroy(&(s->mutex));
	pthread_cond_destroy(&(s->cond));
}

void
pthread_sema_up(pthread_sema_t *s)
{
	int status = pthread_mutex_lock(&(s->mutex));

	if (s->cnt++ < 0) {
		/* wackup sleeping thread */
		status = pthread_cond_signal(&(s->cond));
	}
	status = pthread_mutex_unlock(&(s->mutex));
}

void
pthread_sema_down(pthread_sema_t *s)
{
	int status = pthread_mutex_lock(&(s->mutex));

	if (--s->cnt < 0) {
		/* thread goes to sleep */
		status = pthread_cond_wait(&(s->cond), &(s->mutex));
	}
	status = pthread_mutex_unlock(&(s->mutex));
}
#endif

MT_Id
MT_getpid(void)
{
#ifdef PTW32
	return (MT_Id) (((size_t) pthread_self().p) + 1);
#else
	return (MT_Id) (((size_t) pthread_self()) + 1);
#endif
}

int
MT_alive(int pid)
{
#ifdef HAVE_KILL
	return kill(pid, 0) == 0;
#else
	(void)pid;
	return 0;
#endif
}
@}