LCOV - code coverage report
Current view: top level - gdk - gdk_system.c (source / functions) Hit Total Coverage
Test: coverage.info Lines: 164 218 75.2 %
Date: 2021-10-13 02:24:04 Functions: 22 26 84.6 %

          Line data    Source code
       1             : /*
       2             :  * This Source Code Form is subject to the terms of the Mozilla Public
       3             :  * License, v. 2.0.  If a copy of the MPL was not distributed with this
       4             :  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
       5             :  *
       6             :  * Copyright 1997 - July 2008 CWI, August 2008 - 2021 MonetDB B.V.
       7             :  */
       8             : 
       9             : /*
      10             :  * @a Niels Nes, Peter Boncz
      11             :  * @+ Threads
      12             :  * This file contains a wrapper layer for threading, hence the
      13             :  * underscore convention MT_x (Multi-Threading).  As all platforms
      14             :  * that MonetDB runs on now support POSIX Threads (pthreads), this
      15             :  * wrapping layer has become rather thin.
      16             :  *
      17             :  * In the late 1990s when multi-threading support was introduced in
      18             :  * MonetDB, pthreads was just emerging as a standard API and not
      19             :  * widely adopted yet.  The earliest MT implementation focused on SGI
      20             :  * Unix and provided multi- threading using multiple processses, and
      21             :  * shared memory.
      22             :  *
      23             :  * One of the relics of this model, namely the need to pre-allocate
      24             :  * locks and semaphores, and consequently a maximum number of them,
      25             :  * has been removed in the latest iteration of this layer.
      26             :  *
      27             :  */
      28             : /*
      29             :  * @- Mthreads Routine implementations
      30             :  */
      31             : #include "monetdb_config.h"
      32             : #include "mstring.h"
      33             : #include "gdk.h"
      34             : #include "gdk_system.h"
      35             : #include "gdk_system_private.h"
      36             : 
      37             : #include <time.h>
      38             : 
      39             : #ifdef HAVE_FTIME
      40             : #include <sys/timeb.h>            /* ftime */
      41             : #endif
      42             : #ifdef HAVE_SYS_TIME_H
      43             : #include <sys/time.h>             /* gettimeofday */
      44             : #endif
      45             : 
      46             : #include <signal.h>
      47             : #include <string.h>               /* for strerror */
      48             : #include <unistd.h>               /* for sysconf symbols */
      49             : 
      50             : #ifdef LOCK_STATS
      51             : 
      52             : ATOMIC_TYPE GDKlockcnt = ATOMIC_VAR_INIT(0);
      53             : ATOMIC_TYPE GDKlockcontentioncnt = ATOMIC_VAR_INIT(0);
      54             : ATOMIC_TYPE GDKlocksleepcnt = ATOMIC_VAR_INIT(0);
      55             : MT_Lock *volatile GDKlocklist = 0;
      56             : ATOMIC_FLAG GDKlocklistlock = ATOMIC_FLAG_INIT;
      57             : 
      58             : /* merge sort of linked list */
      59             : static MT_Lock *
      60             : sortlocklist(MT_Lock *l)
      61             : {
      62             :         MT_Lock *r, *t, *ll = NULL;
      63             : 
      64             :         if (l == NULL || l->next == NULL) {
      65             :                 /* list is trivially sorted (0 or 1 element) */
      66             :                 return l;
      67             :         }
      68             :         /* break list into two (almost) equal pieces:
      69             :         * l is start of "left" list, r of "right" list, ll last
      70             :         * element of "left" list */
      71             :         for (t = r = l; t && t->next; t = t->next->next) {
      72             :                 ll = r;
      73             :                 r = r->next;
      74             :         }
      75             :         ll->next = NULL;     /* break list into two */
      76             :         r->prev = NULL;
      77             :         /* recursively sort both sublists */
      78             :         l = sortlocklist(l);
      79             :         r = sortlocklist(r);
      80             :         /* merge
      81             :          * t is new list, ll is last element of new list, l and r are
      82             :          * start of unprocessed part of left and right lists */
      83             :         t = ll = NULL;
      84             :         while (l && r) {
      85             :                 if (ATOMIC_GET(&l->sleep) < ATOMIC_GET(&r->sleep) ||
      86             :                     (ATOMIC_GET(&l->sleep) == ATOMIC_GET(&r->sleep) &&
      87             :                      (ATOMIC_GET(&l->contention) < ATOMIC_GET(&r->contention) ||
      88             :                       (ATOMIC_GET(&l->contention) == ATOMIC_GET(&r->contention) &&
      89             :                        l->count <= r->count)))) {
      90             :                         /* l is smaller */
      91             :                         if (ll == NULL) {
      92             :                                 assert(t == NULL);
      93             :                                 t = ll = l;
      94             :                         } else {
      95             :                                 ll->next = l;
      96             :                                 l->prev = ll;
      97             :                                 ll = ll->next;
      98             :                         }
      99             :                         l = l->next;
     100             :                 } else {
     101             :                         /* r is smaller */
     102             :                         if (ll == NULL) {
     103             :                                 assert(t == NULL);
     104             :                                 t = ll = r;
     105             :                         } else {
     106             :                                 ll->next = r;
     107             :                                 r->prev = ll;
     108             :                                 ll = ll->next;
     109             :                         }
     110             :                         r = r->next;
     111             :                 }
     112             :         }
     113             :         /* append rest of remaining list */
     114             :         if (l) {
     115             :                 ll->next = l;
     116             :                 l->prev = ll;
     117             :         } else {
     118             :                 ll->next = r;
     119             :                 r->prev = ll;
     120             :         }
     121             :         return t;
     122             : }
     123             : 
     124             : static inline bool
     125             : lock_isset(MT_Lock *l)
     126             : {
     127             :         if (MT_lock_try(l)) {
     128             :                 MT_lock_unset(l);
     129             :                 return false;
     130             :         }
     131             :         return true;
     132             : }
     133             : 
     134             : /* function used for debugging */
     135             : void
     136             : GDKlockstatistics(int what)
     137             : {
     138             :         MT_Lock *l;
     139             :         int n = 0;
     140             : 
     141             :         if (ATOMIC_TAS(&GDKlocklistlock) != 0) {
     142             :                 fprintf(stderr, "GDKlocklistlock is set, so cannot access lock list\n");
     143             :                 return;
     144             :         }
     145             :         if (what == -1) {
     146             :                 for (l = GDKlocklist; l; l = l->next) {
     147             :                         l->count = 0;
     148             :                         ATOMIC_SET(&l->contention, 0);
     149             :                         ATOMIC_SET(&l->sleep, 0);
     150             :                 }
     151             :                 ATOMIC_CLEAR(&GDKlocklistlock);
     152             :                 return;
     153             :         }
     154             :         GDKlocklist = sortlocklist(GDKlocklist);
     155             :         fprintf(stderr, "%-18s\t%s\t%s\t%s\t%s\t%s\t%s\n",
     156             :                 "lock name", "count", "content", "sleep",
     157             :                 "locked", "locker", "thread");
     158             :         for (l = GDKlocklist; l; l = l->next) {
     159             :                 n++;
     160             :                 if (what == 0 ||
     161             :                     (what == 1 && l->count) ||
     162             :                     (what == 2 && ATOMIC_GET(&l->contention)) ||
     163             :                     (what == 3 && lock_isset(l)))
     164             :                         fprintf(stderr, "%-18s\t%zu\t%zu\t%zu\t%s\t%s\t%s\n",
     165             :                                 l->name, l->count,
     166             :                                 (size_t) ATOMIC_GET(&l->contention),
     167             :                                 (size_t) ATOMIC_GET(&l->sleep),
     168             :                                 lock_isset(l) ? "locked" : "",
     169             :                                 l->locker ? l->locker : "",
     170             :                                 l->thread ? l->thread : "");
     171             :         }
     172             :         fprintf(stderr, "Number of locks: %d\n", n);
     173             :         fprintf(stderr, "Total lock count: %zu\n", (size_t) ATOMIC_GET(&GDKlockcnt));
     174             :         fprintf(stderr, "Lock contention:  %zu\n", (size_t) ATOMIC_GET(&GDKlockcontentioncnt));
     175             :         fprintf(stderr, "Lock sleep count: %zu\n", (size_t) ATOMIC_GET(&GDKlocksleepcnt));
     176             :         ATOMIC_CLEAR(&GDKlocklistlock);
     177             : }
     178             : 
     179             : #endif  /* LOCK_STATS */
     180             : 
     181             : #if !defined(HAVE_PTHREAD_H) && defined(WIN32)
     182             : static struct winthread {
     183             :         struct winthread *next;
     184             :         HANDLE hdl;
     185             :         DWORD tid;
     186             :         void (*func) (void *);
     187             :         void *data;
     188             :         MT_Lock *lockwait;      /* lock we're waiting for */
     189             :         MT_Sema *semawait;      /* semaphore we're waiting for */
     190             :         struct winthread *joinwait; /* process we are joining with */
     191             :         const char *working;    /* what we're currently doing */
     192             :         char algorithm[512];    /* the algorithm used in the last operation */
     193             :         size_t algolen;         /* length of string in .algorithm */
     194             :         ATOMIC_TYPE exited;
     195             :         bool detached:1, waiting:1;
     196             :         char threadname[MT_NAME_LEN];
     197             :         QryCtx *qry_ctx;
     198             : } *winthreads = NULL;
     199             : static struct winthread mainthread = {
     200             :         .threadname = "main thread",
     201             :         .exited = ATOMIC_VAR_INIT(0),
     202             : };
     203             : 
     204             : static CRITICAL_SECTION winthread_cs;
     205             : static DWORD threadslot = TLS_OUT_OF_INDEXES;
     206             : 
     207             : void
     208             : dump_threads(void)
     209             : {
     210             :         TRC_DEBUG_IF(THRD) {
     211             :                 EnterCriticalSection(&winthread_cs);
     212             :                 for (struct winthread *w = winthreads; w; w = w->next) {
     213             :                         TRC_DEBUG_ENDIF(THRD, "%s, waiting for %s, working on %.200s\n",
     214             :                                         w->threadname,
     215             :                                         w->lockwait ? w->lockwait->name :
     216             :                                         w->semawait ? w->semawait->name :
     217             :                                         w->joinwait ? w->joinwait->threadname :
     218             :                                         "nothing",
     219             :                                         ATOMIC_GET(&w->exited) ? "exiting" :
     220             :                                         w->working ? w->working : "nothing");
     221             :                 }
     222             :                 LeaveCriticalSection(&winthread_cs);
     223             :         }
     224             : }
     225             : 
     226             : bool
     227             : MT_thread_init(void)
     228             : {
     229             :         if (threadslot == TLS_OUT_OF_INDEXES) {
     230             :                 threadslot = TlsAlloc();
     231             :                 if (threadslot == TLS_OUT_OF_INDEXES) {
     232             :                         GDKwinerror("Creating thread-local slot for thread failed");
     233             :                         return false;
     234             :                 }
     235             :                 mainthread.tid = GetCurrentThreadId();
     236             :                 if (TlsSetValue(threadslot, &mainthread) == 0) {
     237             :                         GDKwinerror("Setting thread-local value failed");
     238             :                         TlsFree(threadslot);
     239             :                         threadslot = TLS_OUT_OF_INDEXES;
     240             :                         return false;
     241             :                 }
     242             :                 InitializeCriticalSection(&winthread_cs);
     243             :         }
     244             :         return true;
     245             : }
     246             : 
     247             : static struct winthread *
     248             : find_winthread(DWORD tid)
     249             : {
     250             :         struct winthread *w;
     251             : 
     252             :         EnterCriticalSection(&winthread_cs);
     253             :         for (w = winthreads; w && w->tid != tid; w = w->next)
     254             :                 ;
     255             :         LeaveCriticalSection(&winthread_cs);
     256             :         return w;
     257             : }
     258             : 
     259             : const char *
     260             : MT_thread_getname(void)
     261             : {
     262             :         if (threadslot == TLS_OUT_OF_INDEXES)
     263             :                 return mainthread.threadname;
     264             :         struct winthread *w = TlsGetValue(threadslot);
     265             :         return w ? w->threadname : UNKNOWN_THREAD;
     266             : }
     267             : 
     268             : void
     269             : MT_thread_setdata(void *data)
     270             : {
     271             :         if (threadslot == TLS_OUT_OF_INDEXES)
     272             :                 return;
     273             :         struct winthread *w = TlsGetValue(threadslot);
     274             : 
     275             :         if (w)
     276             :                 w->data = data;
     277             : }
     278             : 
     279             : void *
     280             : MT_thread_getdata(void)
     281             : {
     282             :         if (threadslot == TLS_OUT_OF_INDEXES)
     283             :                 return NULL;
     284             :         struct winthread *w = TlsGetValue(threadslot);
     285             : 
     286             :         return w ? w->data : NULL;
     287             : }
     288             : 
     289             : void
     290             : MT_thread_set_qry_ctx(QryCtx *ctx)
     291             : {
     292             :         if (threadslot == TLS_OUT_OF_INDEXES)
     293             :                 return;
     294             :         struct winthread *w = TlsGetValue(threadslot);
     295             : 
     296             :         if (w)
     297             :                 w->qry_ctx = ctx;
     298             : }
     299             : 
     300             : QryCtx *
     301             : MT_thread_get_qry_ctx(void)
     302             : {
     303             :         if (threadslot == TLS_OUT_OF_INDEXES)
     304             :                 return NULL;
     305             :         struct winthread *w = TlsGetValue(threadslot);
     306             : 
     307             :         return w ? w->qry_ctx : NULL;
     308             : }
     309             : 
     310             : void
     311             : MT_thread_setlockwait(MT_Lock *lock)
     312             : {
     313             :         if (threadslot == TLS_OUT_OF_INDEXES)
     314             :                 return;
     315             :         struct winthread *w = TlsGetValue(threadslot);
     316             : 
     317             :         if (w)
     318             :                 w->lockwait = lock;
     319             : }
     320             : 
     321             : void
     322             : MT_thread_setsemawait(MT_Sema *sema)
     323             : {
     324             :         if (threadslot == TLS_OUT_OF_INDEXES)
     325             :                 return;
     326             :         struct winthread *w = TlsGetValue(threadslot);
     327             : 
     328             :         if (w)
     329             :                 w->semawait = sema;
     330             : }
     331             : 
     332             : void
     333             : MT_thread_setworking(const char *work)
     334             : {
     335             :         if (threadslot == TLS_OUT_OF_INDEXES)
     336             :                 return;
     337             :         struct winthread *w = TlsGetValue(threadslot);
     338             : 
     339             :         if (w)
     340             :                 w->working = work;
     341             : }
     342             : 
     343             : void
     344             : MT_thread_setalgorithm(const char *algo)
     345             : {
     346             :         if (threadslot == TLS_OUT_OF_INDEXES)
     347             :                 return;
     348             :         struct winthread *w = TlsGetValue(threadslot);
     349             : 
     350             :         if (w) {
     351             :                 if (algo) {
     352             :                         if (w->algolen > 0) {
     353             :                                 if (w->algolen < sizeof(w->algorithm))
     354             :                                         w->algolen += strconcat_len(w->algorithm + w->algolen, sizeof(w->algorithm) - w->algolen, "; ", algo, NULL);
     355             :                         } else
     356             :                                 w->algolen = strcpy_len(w->algorithm, algo, sizeof(w->algorithm));
     357             :                 } else {
     358             :                         w->algorithm[0] = 0;
     359             :                         w->algolen = 0;
     360             :                 }
     361             :         }
     362             : }
     363             : 
     364             : const char *
     365             : MT_thread_getalgorithm(void)
     366             : {
     367             :         if (threadslot == TLS_OUT_OF_INDEXES)
     368             :                 return NULL;
     369             :         struct winthread *w = TlsGetValue(threadslot);
     370             : 
     371             :         return w && w->algorithm[0] ? w->algorithm : NULL;
     372             : }
     373             : 
     374             : bool
     375             : MT_thread_override_limits(void)
     376             : {
     377             :         if (threadslot == TLS_OUT_OF_INDEXES)
     378             :                 return false;
     379             :         struct winthread *w = TlsGetValue(threadslot);
     380             : 
     381             :         return w && w->working && strcmp(w->working, "store locked") == 0;
     382             : }
     383             : 
     384             : static void
     385             : rm_winthread(struct winthread *w)
     386             : {
     387             :         struct winthread **wp;
     388             : 
     389             :         EnterCriticalSection(&winthread_cs);
     390             :         for (wp = &winthreads; *wp && *wp != w; wp = &(*wp)->next)
     391             :                 ;
     392             :         if (*wp)
     393             :                 *wp = w->next;
     394             :         LeaveCriticalSection(&winthread_cs);
     395             :         ATOMIC_DESTROY(&w->exited);
     396             :         free(w);
     397             : }
     398             : 
     399             : static DWORD WINAPI
     400             : thread_starter(LPVOID arg)
     401             : {
     402             :         struct winthread *w = (struct winthread *) arg;
     403             :         void *data = w->data;
     404             : 
     405             :         w->data = NULL;
     406             :         TlsSetValue(threadslot, w);
     407             :         (*w->func)(data);
     408             :         ATOMIC_SET(&w->exited, 1);
     409             :         TRC_DEBUG(THRD, "Exit: \"%s\"\n", w->threadname);
     410             :         return 0;
     411             : }
     412             : 
     413             : static void
     414             : join_threads(void)
     415             : {
     416             :         bool waited;
     417             : 
     418             :         struct winthread *self = TlsGetValue(threadslot);
     419             :         if (!self)
     420             :                 return;
     421             :         EnterCriticalSection(&winthread_cs);
     422             :         do {
     423             :                 waited = false;
     424             :                 for (struct winthread *w = winthreads; w; w = w->next) {
     425             :                         if (w->detached && !w->waiting && ATOMIC_GET(&w->exited)) {
     426             :                                 w->waiting = true;
     427             :                                 LeaveCriticalSection(&winthread_cs);
     428             :                                 TRC_DEBUG(THRD, "Join thread \"%s\"\n", w->threadname);
     429             :                                 self->joinwait = w;
     430             :                                 WaitForSingleObject(w->hdl, INFINITE);
     431             :                                 self->joinwait = NULL;
     432             :                                 CloseHandle(w->hdl);
     433             :                                 rm_winthread(w);
     434             :                                 waited = true;
     435             :                                 EnterCriticalSection(&winthread_cs);
     436             :                                 break;
     437             :                         }
     438             :                 }
     439             :         } while (waited);
     440             :         LeaveCriticalSection(&winthread_cs);
     441             : }
     442             : 
     443             : void
     444             : join_detached_threads(void)
     445             : {
     446             :         bool waited;
     447             : 
     448             :         struct winthread *self = TlsGetValue(threadslot);
     449             :         EnterCriticalSection(&winthread_cs);
     450             :         do {
     451             :                 waited = false;
     452             :                 for (struct winthread *w = winthreads; w; w = w->next) {
     453             :                         if (w->detached && !w->waiting) {
     454             :                                 w->waiting = true;
     455             :                                 LeaveCriticalSection(&winthread_cs);
     456             :                                 TRC_DEBUG(THRD, "Join thread \"%s\"\n", w->threadname);
     457             :                                 self->joinwait = w;
     458             :                                 WaitForSingleObject(w->hdl, INFINITE);
     459             :                                 self->joinwait = NULL;
     460             :                                 CloseHandle(w->hdl);
     461             :                                 rm_winthread(w);
     462             :                                 waited = true;
     463             :                                 EnterCriticalSection(&winthread_cs);
     464             :                                 break;
     465             :                         }
     466             :                 }
     467             :         } while (waited);
     468             :         LeaveCriticalSection(&winthread_cs);
     469             : }
     470             : 
     471             : int
     472             : MT_create_thread(MT_Id *t, void (*f) (void *), void *arg, enum MT_thr_detach d, const char *threadname)
     473             : {
     474             :         struct winthread *w;
     475             : 
     476             :         join_threads();
     477             :         if (threadname == NULL) {
     478             :                 TRC_CRITICAL(GDK, "Thread must have a name\n");
     479             :                 return -1;
     480             :         }
     481             :         if (strlen(threadname) >= sizeof(w->threadname)) {
     482             :                 TRC_CRITICAL(GDK, "Thread's name is too large\n");
     483             :                 return -1;
     484             :         }
     485             : 
     486             :         w = malloc(sizeof(*w));
     487             :         if (w == NULL) {
     488             :                 GDKsyserror("Cannot allocate memory\n");
     489             :                 return -1;
     490             :         }
     491             : 
     492             :         *w = (struct winthread) {
     493             :                 .func = f,
     494             :                 .data = arg,
     495             :                 .waiting = false,
     496             :                 .detached = (d == MT_THR_DETACHED),
     497             :         };
     498             :         ATOMIC_INIT(&w->exited, 0);
     499             :         strcpy_len(w->threadname, threadname, sizeof(w->threadname));
     500             :         TRC_DEBUG(THRD, "Create thread \"%s\"\n", threadname);
     501             :         EnterCriticalSection(&winthread_cs);
     502             :         w->hdl = CreateThread(NULL, THREAD_STACK_SIZE, thread_starter, w,
     503             :                               0, &w->tid);
     504             :         if (w->hdl == NULL) {
     505             :                 GDKwinerror("Failed to create thread");
     506             :                 LeaveCriticalSection(&winthread_cs);
     507             :                 free(w);
     508             :                 return -1;
     509             :         }
     510             :         /* must not fail after this: the thread has been started */
     511             :         w->next = winthreads;
     512             :         winthreads = w;
     513             :         LeaveCriticalSection(&winthread_cs);
     514             :         *t = (MT_Id) w->tid;
     515             :         return 0;
     516             : }
     517             : 
     518             : MT_Id
     519             : MT_getpid(void)
     520             : {
     521             :         return (MT_Id) GetCurrentThreadId();
     522             : }
     523             : 
     524             : void
     525             : MT_exiting_thread(void)
     526             : {
     527             :         if (threadslot == TLS_OUT_OF_INDEXES)
     528             :                 return;
     529             : 
     530             :         struct winthread *w = TlsGetValue(threadslot);
     531             : 
     532             :         if (w) {
     533             :                 ATOMIC_SET(&w->exited, 1);
     534             :                 w->working = NULL;
     535             :         }
     536             : }
     537             : 
     538             : int
     539             : MT_join_thread(MT_Id t)
     540             : {
     541             :         struct winthread *w;
     542             : 
     543             :         assert(t != mainthread.tid);
     544             :         join_threads();
     545             :         w = find_winthread((DWORD) t);
     546             :         if (w == NULL || w->hdl == NULL)
     547             :                 return -1;
     548             :         TRC_DEBUG(THRD, "Join thread \"%s\"\n", w->threadname);
     549             :         struct winthread *self = TlsGetValue(threadslot);
     550             :         self->joinwait = w;
     551             :         DWORD ret = WaitForSingleObject(w->hdl, INFINITE);
     552             :         self->joinwait = NULL;
     553             :         if (ret == WAIT_OBJECT_0 && CloseHandle(w->hdl)) {
     554             :                 rm_winthread(w);
     555             :                 return 0;
     556             :         }
     557             :         return -1;
     558             : }
     559             : 
     560             : int
     561             : MT_kill_thread(MT_Id t)
     562             : {
     563             :         struct winthread *w;
     564             : 
     565             :         assert(t != mainthread.tid);
     566             :         join_threads();
     567             :         w = find_winthread((DWORD) t);
     568             :         if (w == NULL)
     569             :                 return -1;
     570             :         if (w->hdl == NULL) {
     571             :                 /* detached thread */
     572             :                 HANDLE h;
     573             :                 int ret = 0;
     574             :                 h = OpenThread(THREAD_ALL_ACCESS, 0, (DWORD) t);
     575             :                 if (h == NULL)
     576             :                         return -1;
     577             :                 if (TerminateThread(h, -1))
     578             :                         ret = -1;
     579             :                 CloseHandle(h);
     580             :                 return ret;
     581             :         }
     582             :         if (TerminateThread(w->hdl, -1))
     583             :                 return 0;
     584             :         return -1;
     585             : }
     586             : 
     587             : #else  /* !defined(HAVE_PTHREAD_H) && defined(_MSC_VER) */
     588             : 
     589             : static struct posthread {
     590             :         struct posthread *next;
     591             :         void (*func)(void *);
     592             :         void *data;
     593             :         MT_Lock *lockwait;      /* lock we're waiting for */
     594             :         MT_Sema *semawait;      /* semaphore we're waiting for */
     595             :         struct posthread *joinwait; /* process we are joining with */
     596             :         const char *working;    /* what we're currently doing */
     597             :         char algorithm[512];    /* the algorithm used in the last operation */
     598             :         size_t algolen;         /* length of string in .algorithm */
     599             :         char threadname[MT_NAME_LEN];
     600             :         pthread_t tid;
     601             :         MT_Id mtid;
     602             :         ATOMIC_TYPE exited;
     603             :         bool detached:1, waiting:1;
     604             :         QryCtx *qry_ctx;
     605             : } *posthreads = NULL;
     606             : static struct posthread mainthread = {
     607             :         .threadname = "main thread",
     608             :         .mtid = 1,
     609             :         .exited = ATOMIC_VAR_INIT(0),
     610             : };
     611             : static pthread_mutex_t posthread_lock = PTHREAD_MUTEX_INITIALIZER;
     612             : static MT_Id MT_thread_id = 1;
     613             : 
     614             : static pthread_key_t threadkey;
     615             : static bool thread_initialized = false;
     616             : 
     617             : void
     618           0 : dump_threads(void)
     619             : {
     620           0 :         TRC_DEBUG_IF(THRD) {
     621           0 :                 pthread_mutex_lock(&posthread_lock);
     622           0 :                 for (struct posthread *p = posthreads; p; p = p->next) {
     623           0 :                         TRC_DEBUG_ENDIF(THRD, "%s, waiting for %s, working on %.200s\n",
     624             :                                         p->threadname,
     625             :                                         p->lockwait ? p->lockwait->name :
     626             :                                         p->semawait ? p->semawait->name :
     627             :                                         p->joinwait ? p->joinwait->threadname :
     628             :                                         "nothing",
     629             :                                         ATOMIC_GET(&p->exited) ? "exiting" :
     630             :                                         p->working ? p->working : "nothing");
     631             :                 }
     632           0 :                 pthread_mutex_unlock(&posthread_lock);
     633             :         }
     634           0 : }
     635             : 
     636             : bool
     637         258 : MT_thread_init(void)
     638             : {
     639             :         int ret;
     640             : 
     641         258 :         if ((ret = pthread_key_create(&threadkey, NULL)) != 0) {
     642           0 :                 GDKsyserr(ret, "Creating specific key for thread failed");
     643             :                 return false;
     644             :         }
     645         258 :         thread_initialized = true;
     646         258 :         mainthread.tid = pthread_self();
     647         258 :         if ((ret = pthread_setspecific(threadkey, &mainthread)) != 0) {
     648           0 :                 GDKsyserr(ret, "Setting specific value failed");
     649             :                 return false;
     650             :         }
     651             :         return true;
     652             : }
     653             : 
     654             : static struct posthread *
     655        3635 : find_posthread(MT_Id tid)
     656             : {
     657             :         struct posthread *p;
     658             : 
     659        3635 :         pthread_mutex_lock(&posthread_lock);
     660        6209 :         for (p = posthreads; p && p->mtid != tid; p = p->next)
     661             :                 ;
     662        3635 :         pthread_mutex_unlock(&posthread_lock);
     663        3635 :         return p;
     664             : }
     665             : 
     666             : const char *
     667       17356 : MT_thread_getname(void)
     668             : {
     669             :         struct posthread *p;
     670             : 
     671       17356 :         if (!thread_initialized)
     672             :                 return mainthread.threadname;
     673       17356 :         p = pthread_getspecific(threadkey);
     674       17356 :         return p ? p->threadname : UNKNOWN_THREAD;
     675             : }
     676             : 
     677             : void
     678       19071 : MT_thread_setdata(void *data)
     679             : {
     680       19071 :         if (!thread_initialized)
     681             :                 return;
     682       19071 :         struct posthread *p = pthread_getspecific(threadkey);
     683             : 
     684       19069 :         if (p)
     685       19069 :                 p->data = data;
     686             : }
     687             : 
     688             : void *
     689   195897249 : MT_thread_getdata(void)
     690             : {
     691   195897249 :         if (!thread_initialized)
     692             :                 return NULL;
     693   195897249 :         struct posthread *p = pthread_getspecific(threadkey);
     694             : 
     695   195876650 :         return p ? p->data : NULL;
     696             : }
     697             : 
     698             : void
     699    24031774 : MT_thread_set_qry_ctx(QryCtx *ctx)
     700             : {
     701    24031774 :         if (!thread_initialized)
     702             :                 return;
     703    24031948 :         struct posthread *p = pthread_getspecific(threadkey);
     704             : 
     705    24031112 :         if (p)
     706    24031112 :                 p->qry_ctx = ctx;
     707             : }
     708             : 
     709             : QryCtx *
     710    18961114 : MT_thread_get_qry_ctx(void)
     711             : {
     712    18961114 :         if (!thread_initialized)
     713             :                 return NULL;
     714    18961114 :         struct posthread *p = pthread_getspecific(threadkey);
     715             : 
     716    18961135 :         return p ? p->qry_ctx : NULL;
     717             : }
     718             : 
     719             : void
     720           0 : MT_thread_setlockwait(MT_Lock *lock)
     721             : {
     722           0 :         if (!thread_initialized)
     723             :                 return;
     724           0 :         struct posthread *p = pthread_getspecific(threadkey);
     725             : 
     726           0 :         if (p)
     727           0 :                 p->lockwait = lock;
     728             : }
     729             : 
     730             : void
     731    16126084 : MT_thread_setsemawait(MT_Sema *sema)
     732             : {
     733    16126084 :         if (!thread_initialized)
     734             :                 return;
     735    16126470 :         struct posthread *p = pthread_getspecific(threadkey);
     736             : 
     737    16126249 :         if (p)
     738    16126249 :                 p->semawait = sema;
     739             : }
     740             : 
     741             : void
     742    14819529 : MT_thread_setworking(const char *work)
     743             : {
     744    14819529 :         if (!thread_initialized)
     745             :                 return;
     746    14819558 :         struct posthread *p = pthread_getspecific(threadkey);
     747             : 
     748    14818422 :         if (p)
     749    14818422 :                 p->working = work;
     750             : }
     751             : 
     752             : void
     753    62638490 : MT_thread_setalgorithm(const char *algo)
     754             : {
     755    62638490 :         if (!thread_initialized)
     756             :                 return;
     757    62638556 :         struct posthread *p = pthread_getspecific(threadkey);
     758             : 
     759    62639360 :         if (p) {
     760    62639360 :                 if (algo) {
     761     3634491 :                         if (p->algolen > 0) {
     762     1805578 :                                 if (p->algolen < sizeof(p->algorithm))
     763     1132096 :                                         p->algolen += strconcat_len(p->algorithm + p->algolen, sizeof(p->algorithm) - p->algolen, "; ", algo, NULL);
     764             :                         } else
     765     1828913 :                                 p->algolen = strcpy_len(p->algorithm, algo, sizeof(p->algorithm));
     766             :                 } else {
     767    59004869 :                         p->algorithm[0] = 0;
     768    59004869 :                         p->algolen = 0;
     769             :                 }
     770             :         }
     771             : }
     772             : 
     773             : const char *
     774        1653 : MT_thread_getalgorithm(void)
     775             : {
     776        1653 :         if (!thread_initialized)
     777             :                 return NULL;
     778        1653 :         struct posthread *p = pthread_getspecific(threadkey);
     779             : 
     780        1653 :         return p && p->algorithm[0] ? p->algorithm : NULL;
     781             : }
     782             : 
     783             : bool
     784           0 : MT_thread_override_limits(void)
     785             : {
     786           0 :         if (!thread_initialized)
     787             :                 return false;
     788           0 :         struct posthread *p = pthread_getspecific(threadkey);
     789             : 
     790           0 :         return p && p->working && strcmp(p->working, "store locked") == 0;
     791             : }
     792             : 
     793             : #ifdef HAVE_PTHREAD_SIGMASK
     794             : static void
     795       38030 : MT_thread_sigmask(sigset_t *new_mask, sigset_t *orig_mask)
     796             : {
     797             :         /* do not check for errors! */
     798       38030 :         sigdelset(new_mask, SIGQUIT);
     799       38030 :         sigdelset(new_mask, SIGPROF);
     800       38030 :         pthread_sigmask(SIG_SETMASK, new_mask, orig_mask);
     801       38030 : }
     802             : #endif
     803             : 
     804             : static void
     805       19005 : rm_posthread_locked(struct posthread *p)
     806             : {
     807             :         struct posthread **pp;
     808             : 
     809       26609 :         for (pp = &posthreads; *pp && *pp != p; pp = &(*pp)->next)
     810             :                 ;
     811       19005 :         if (*pp)
     812       19005 :                 *pp = p->next;
     813             :         ATOMIC_DESTROY(&p->exited);
     814       19005 :         free(p);
     815       19005 : }
     816             : 
     817             : static void
     818       19005 : rm_posthread(struct posthread *p)
     819             : {
     820       19005 :         pthread_mutex_lock(&posthread_lock);
     821       19005 :         rm_posthread_locked(p);
     822       19005 :         pthread_mutex_unlock(&posthread_lock);
     823       19005 : }
     824             : 
     825             : static void *
     826       19015 : thread_starter(void *arg)
     827             : {
     828             :         struct posthread *p = (struct posthread *) arg;
     829       19015 :         void *data = p->data;
     830             : 
     831       19015 :         p->data = NULL;
     832       19015 :         pthread_setspecific(threadkey, p);
     833       19015 :         (*p->func)(data);
     834       19009 :         ATOMIC_SET(&p->exited, 1);
     835       19009 :         TRC_DEBUG(THRD, "Exit thread \"%s\"\n", p->threadname);
     836       19009 :         return NULL;
     837             : }
     838             : 
     839             : static void
     840       22650 : join_threads(void)
     841             : {
     842             :         bool waited;
     843             : 
     844       22650 :         struct posthread *self = pthread_getspecific(threadkey);
     845       22650 :         pthread_mutex_lock(&posthread_lock);
     846             :         do {
     847             :                 waited = false;
     848      238066 :                 for (struct posthread *p = posthreads; p; p = p->next) {
     849      215416 :                         if (p->detached && !p->waiting && ATOMIC_GET(&p->exited)) {
     850       14150 :                                 p->waiting = true;
     851       14150 :                                 pthread_mutex_unlock(&posthread_lock);
     852       14150 :                                 TRC_DEBUG(THRD, "Join thread \"%s\"\n", p->threadname);
     853       14150 :                                 if (self) self->joinwait = p;
     854       14150 :                                 pthread_join(p->tid, NULL);
     855       14150 :                                 if (self) self->joinwait = NULL;
     856       14150 :                                 rm_posthread(p);
     857             :                                 waited = true;
     858       14150 :                                 pthread_mutex_lock(&posthread_lock);
     859             :                                 break;
     860             :                         }
     861             :                 }
     862             :         } while (waited);
     863       22650 :         pthread_mutex_unlock(&posthread_lock);
     864       22650 : }
     865             : 
     866             : void
     867         528 : join_detached_threads(void)
     868             : {
     869             :         bool waited;
     870             : 
     871         528 :         struct posthread *self = pthread_getspecific(threadkey);
     872         528 :         pthread_mutex_lock(&posthread_lock);
     873             :         do {
     874             :                 waited = false;
     875        6040 :                 for (struct posthread *p = posthreads; p; p = p->next) {
     876        5512 :                         if (p->detached && !p->waiting) {
     877        1220 :                                 p->waiting = true;
     878        1220 :                                 pthread_mutex_unlock(&posthread_lock);
     879        1220 :                                 TRC_DEBUG(THRD, "Join thread \"%s\"\n", p->threadname);
     880        1220 :                                 if (self) self->joinwait = p;
     881        1220 :                                 pthread_join(p->tid, NULL);
     882        1220 :                                 if (self) self->joinwait = NULL;
     883        1220 :                                 rm_posthread(p);
     884             :                                 waited = true;
     885        1220 :                                 pthread_mutex_lock(&posthread_lock);
     886             :                                 break;
     887             :                         }
     888             :                 }
     889             :         } while (waited);
     890         528 :         pthread_mutex_unlock(&posthread_lock);
     891         528 : }
     892             : 
     893             : int
     894       19015 : MT_create_thread(MT_Id *t, void (*f) (void *), void *arg, enum MT_thr_detach d, const char *threadname)
     895             : {
     896             :         pthread_attr_t attr;
     897             :         int ret;
     898             :         struct posthread *p;
     899             : 
     900       19015 :         join_threads();
     901       19015 :         if (threadname == NULL) {
     902           0 :                 TRC_CRITICAL(GDK, "Thread must have a name\n");
     903           0 :                 return -1;
     904             :         }
     905       19015 :         if (strlen(threadname) >= sizeof(p->threadname)) {
     906           0 :                 TRC_CRITICAL(GDK, "Thread's name is too large\n");
     907           0 :                 return -1;
     908             :         }
     909       19015 :         if ((ret = pthread_attr_init(&attr)) != 0) {
     910           0 :                 GDKsyserr(ret, "Cannot init pthread attr");
     911             :                 return -1;
     912             :         }
     913       19015 :         if ((ret = pthread_attr_setstacksize(&attr, THREAD_STACK_SIZE)) != 0) {
     914           0 :                 GDKsyserr(ret, "Cannot set stack size");
     915           0 :                 pthread_attr_destroy(&attr);
     916             :                 return -1;
     917             :         }
     918       19015 :         p = malloc(sizeof(struct posthread));
     919       19015 :         if (p == NULL) {
     920           0 :                 GDKsyserror("Cannot allocate memory\n");
     921           0 :                 pthread_attr_destroy(&attr);
     922             :                 return -1;
     923             :         }
     924       19015 :         *p = (struct posthread) {
     925             :                 .func = f,
     926             :                 .data = arg,
     927             :                 .waiting = false,
     928       19015 :                 .detached = (d == MT_THR_DETACHED),
     929             :         };
     930       19015 :         ATOMIC_INIT(&p->exited, 0);
     931             : 
     932       19015 :         strcpy_len(p->threadname, threadname, sizeof(p->threadname));
     933             : #ifdef HAVE_PTHREAD_SIGMASK
     934             :         sigset_t new_mask, orig_mask;
     935       19015 :         (void) sigfillset(&new_mask);
     936       19015 :         MT_thread_sigmask(&new_mask, &orig_mask);
     937             : #endif
     938       19015 :         TRC_DEBUG(THRD, "Create thread \"%s\"\n", threadname);
     939             :         /* protect posthreads during thread creation and only add to
     940             :          * it after the thread was created successfully */
     941       19015 :         pthread_mutex_lock(&posthread_lock);
     942       19015 :         *t = p->mtid = ++MT_thread_id;
     943       19015 :         ret = pthread_create(&p->tid, &attr, thread_starter, p);
     944       19015 :         if (ret != 0) {
     945           0 :                 pthread_mutex_unlock(&posthread_lock);
     946           0 :                 GDKsyserr(ret, "Cannot start thread");
     947           0 :                 free(p);
     948             :                 ret = -1;
     949             :         } else {
     950             :                 /* must not fail after this: the thread has been started */
     951       19015 :                 p->next = posthreads;
     952       19015 :                 posthreads = p;
     953       19015 :                 pthread_mutex_unlock(&posthread_lock);
     954             :         }
     955       19015 :         (void) pthread_attr_destroy(&attr); /* not interested in errors */
     956             : #ifdef HAVE_PTHREAD_SIGMASK
     957       19015 :         MT_thread_sigmask(&orig_mask, NULL);
     958             : #endif
     959       19015 :         return ret;
     960             : }
     961             : 
     962             : MT_Id
     963    80090322 : MT_getpid(void)
     964             : {
     965             :         struct posthread *p;
     966             : 
     967    80090322 :         if (!thread_initialized)
     968         258 :                 return mainthread.mtid;
     969    80090064 :         p = pthread_getspecific(threadkey);
     970    80073867 :         return p ? p->mtid : 0;
     971             : }
     972             : 
     973             : void
     974        5367 : MT_exiting_thread(void)
     975             : {
     976             :         struct posthread *p;
     977             : 
     978        5367 :         if (!thread_initialized)
     979             :                 return;
     980        5367 :         p = pthread_getspecific(threadkey);
     981        5367 :         if (p) {
     982        5367 :                 ATOMIC_SET(&p->exited, 1);
     983        5367 :                 p->working = NULL;
     984             :         }
     985             : }
     986             : 
     987             : int
     988        3635 : MT_join_thread(MT_Id t)
     989             : {
     990             :         struct posthread *p;
     991             :         int ret;
     992             : 
     993        3635 :         assert(t > 1);
     994        3635 :         join_threads();
     995        3635 :         p = find_posthread(t);
     996        3635 :         if (p == NULL)
     997             :                 return -1;
     998        3635 :         TRC_DEBUG(THRD, "Join thread \"%s\"\n", p->threadname);
     999        3635 :         struct posthread *self = pthread_getspecific(threadkey);
    1000        3635 :         if (self) self->joinwait = p;
    1001        3635 :         ret = pthread_join(p->tid, NULL);
    1002        3635 :         if (self) self->joinwait = NULL;
    1003        3635 :         if (ret != 0) {
    1004           0 :                 GDKsyserr(ret, "Joining thread failed");
    1005             :                 return -1;
    1006             :         }
    1007        3635 :         rm_posthread(p);
    1008        3635 :         return 0;
    1009             : }
    1010             : 
    1011             : int
    1012           0 : MT_kill_thread(MT_Id t)
    1013             : {
    1014           0 :         assert(t > 1);
    1015             : #ifdef HAVE_PTHREAD_KILL
    1016             :         struct posthread *p;
    1017             : 
    1018           0 :         join_threads();
    1019           0 :         p = find_posthread(t);
    1020           0 :         if (p)
    1021           0 :                 return pthread_kill(p->tid, SIGHUP);
    1022             : #else
    1023             :         (void) t;
    1024             :         join_threads();
    1025             : #endif
    1026             :         return -1;
    1027             : }
    1028             : #endif
    1029             : 
    1030             : int
    1031         266 : MT_check_nr_cores(void)
    1032             : {
    1033             :         int ncpus = -1;
    1034             : 
    1035             : #if defined(HAVE_SYSCONF) && defined(_SC_NPROCESSORS_ONLN)
    1036             :         /* this works on Linux, Solaris and AIX */
    1037         266 :         ncpus = sysconf(_SC_NPROCESSORS_ONLN);
    1038             : #elif defined(HW_NCPU)   /* BSD */
    1039             :         size_t len = sizeof(int);
    1040             :         int mib[3];
    1041             : 
    1042             :         /* Everyone should have permission to make this call,
    1043             :          * if we get a failure something is really wrong. */
    1044             :         mib[0] = CTL_HW;
    1045             :         mib[1] = HW_NCPU;
    1046             :         mib[2] = -1;
    1047             :         sysctl(mib, 3, &ncpus, &len, NULL, 0);
    1048             : #elif defined(WIN32)
    1049             :         SYSTEM_INFO sysinfo;
    1050             : 
    1051             :         GetSystemInfo(&sysinfo);
    1052             :         ncpus = sysinfo.dwNumberOfProcessors;
    1053             : #endif
    1054             : 
    1055             :         /* if we ever need HPUX or OSF/1 (hope not), see
    1056             :          * http://ndevilla.free.fr/threads/ */
    1057             : 
    1058             :         if (ncpus <= 0)
    1059             :                 ncpus = 1;
    1060             : #if SIZEOF_SIZE_T == SIZEOF_INT
    1061             :         /* On 32-bits systems with large numbers of cpus/cores, we
    1062             :          * quickly run out of space due to the number of threads in
    1063             :          * use.  Since it is questionable whether many cores on a
    1064             :          * 32-bits system are going to be beneficial due to this, we
    1065             :          * simply limit the auto-detected cores to 16 on 32-bits
    1066             :          * systems.  The user can always override this via
    1067             :          * gdk_nr_threads. */
    1068             :         if (ncpus > 16)
    1069             :                 ncpus = 16;
    1070             : #endif
    1071             : 
    1072             : #ifndef WIN32
    1073             :         /* get the number of allocated cpus from the cgroup settings */
    1074         266 :         FILE *f = fopen("/sys/fs/cgroup/cpuset/cpuset.cpus", "r");
    1075         266 :         if (f != NULL) {
    1076             :                 char buf[512];
    1077           0 :                 char *p = fgets(buf, 512, f);
    1078           0 :                 fclose(f);
    1079           0 :                 if (p != NULL) {
    1080             :                         /* syntax is: ranges of CPU numbers separated
    1081             :                          * by comma; a range is either a single CPU
    1082             :                          * id, or two IDs separated by a minus; any
    1083             :                          * deviation causes the file to be ignored */
    1084             :                         int ncpu = 0;
    1085           0 :                         for (;;) {
    1086             :                                 char *q;
    1087           0 :                                 unsigned fst = strtoul(p, &q, 10);
    1088           0 :                                 if (q == p)
    1089           0 :                                         return ncpus;
    1090           0 :                                 ncpu++;
    1091           0 :                                 if (*q == '-') {
    1092           0 :                                         p = q + 1;
    1093           0 :                                         unsigned lst = strtoul(p, &q, 10);
    1094           0 :                                         if (q == p || lst <= fst)
    1095             :                                                 return ncpus;
    1096           0 :                                         ncpu += lst - fst;
    1097             :                                 }
    1098           0 :                                 if (*q == '\n')
    1099             :                                         break;
    1100           0 :                                 if (*q != ',')
    1101             :                                         return ncpus;
    1102           0 :                                 p = q + 1;
    1103             :                         }
    1104           0 :                         if (ncpu < ncpus)
    1105             :                                 return ncpu;
    1106             :                 }
    1107             :         }
    1108             : #endif
    1109             : 
    1110             :         return ncpus;
    1111             : }

Generated by: LCOV version 1.14