source: libcfa/src/concurrency/thread.cfa @ 257a8f5

ADTast-experimental
Last change on this file since 257a8f5 was f5f2768, checked in by Peter A. Buhr <pabuhr@…>, 21 months ago

make _GNU_SOURCE default, change IO to use SOCKADDR_ARG and CONST_SOCKADDR_ARG, move sys/socket.h to first include because of anonymous naming problem

  • Property mode set to 100644
File size: 7.2 KB
Line 
1//
2// Cforall Version 1.0.0 Copyright (C) 2016 University of Waterloo
3//
4// The contents of this file are covered under the licence agreement in the
5// file "LICENCE" distributed with Cforall.
6//
7// thread.c --
8//
9// Author           : Thierry Delisle
10// Created On       : Tue Jan 17 12:27:26 2017
11// Last Modified By : Peter A. Buhr
12// Last Modified On : Mon Jan  9 08:42:33 2023
13// Update Count     : 103
14//
15
16#define __cforall_thread__
17
18#include "thread.hfa"
19
20#include "exception.hfa"
21#include "kernel/private.hfa"
22#include "limits.hfa"
23
24#define __CFA_INVOKE_PRIVATE__
25#include "invoke.h"
26
27extern size_t __global_random_seed;
28extern size_t __global_random_prime;
29extern bool __global_random_mask;
30
31#pragma GCC visibility push(default)
32
33//-----------------------------------------------------------------------------
34// Thread ctors and dtors
35void ?{}( thread$ & this, const char * const name, cluster & cl, void * storage, size_t storageSize ) with( this ) {
36        context{ 0p, 0p };
37        self_cor{ name, storage, storageSize };
38        ticket = TICKET_RUNNING;
39        state = Start;
40        preempted = __NO_PREEMPTION;
41        corctx_flag = false;
42        curr_cor = &self_cor;
43        self_mon.owner = &this;
44        self_mon.recursion = 1;
45        self_mon_p = &self_mon;
46        curr_cluster = &cl;
47        rdy_link.next = 0p;
48        rdy_link.ts   = MAX;
49        user_link.next = 0p;
50        user_link.prev = 0p;
51        cltr_link.next = 0p;
52        cltr_link.prev = 0p;
53        preferred = ready_queue_new_preferred();
54        last_proc = 0p;
55        PRNG_SET_SEED( random_state, __global_random_mask ? __global_random_prime : __global_random_prime ^ rdtscl() );
56        #if defined( __CFA_WITH_VERIFY__ )
57                executing = 0p;
58                canary = 0x0D15EA5E0D15EA5Ep;
59        #endif
60
61        clh_node = malloc( );
62        *clh_node = false;
63
64        doregister(curr_cluster, this);
65        monitors{ &self_mon_p, 1, (fptr_t)0 };
66}
67
68void ^?{}(thread$& this) with( this ) {
69        #if defined( __CFA_WITH_VERIFY__ )
70                canary = 0xDEADDEADDEADDEADp;
71        #endif
72        free(clh_node);
73        unregister(curr_cluster, this);
74        ^self_cor{};
75}
76
77forall(T &)
78void copy(ThreadCancelled(T) * dst, ThreadCancelled(T) * src) {
79        dst->virtual_table = src->virtual_table;
80        dst->the_thread = src->the_thread;
81        dst->the_exception = src->the_exception;
82}
83
84forall(T &)
85const char * msg(ThreadCancelled(T) *) {
86        return "ThreadCancelled(...)";
87}
88
89forall(T &)
90static void default_thread_cancel_handler(ThreadCancelled(T) & ) {
91        // Improve this error message, can I do formatting?
92        abort( "Unhandled thread cancellation.\n" );
93}
94
95forall(T & | is_thread(T) | IS_EXCEPTION(ThreadCancelled(T))
96    | { EHM_DEFAULT_VTABLE(ThreadCancelled(T)); })
97void ?{}( thread_dtor_guard_t & this,
98                T & thrd, void(*cancelHandler)(ThreadCancelled(T) &)) {
99        monitor$ * m = get_monitor(thrd);
100        thread$ * desc = get_thread(thrd);
101
102        // Setup the monitor guard
103        void (*dtor)(T& mutex this) = ^?{};
104        bool join = cancelHandler != (void(*)(ThreadCancelled(T)&))0;
105        (this.mg){&m, (void(*)())dtor, join};
106
107
108        /* paranoid */ verifyf( Halted == desc->state || Cancelled == desc->state, "Expected thread to be Halted or Cancelled, was %d\n", (int)desc->state );
109
110        // After the guard set-up and any wait, check for cancellation.
111        struct _Unwind_Exception * cancellation = desc->self_cor.cancellation;
112        if ( likely( 0p == cancellation ) ) {
113                return;
114        } else if ( Cancelled == desc->state ) {
115                return;
116        }
117        desc->state = Cancelled;
118        void(*defaultResumptionHandler)(ThreadCancelled(T) &) =
119                join ? cancelHandler : default_thread_cancel_handler;
120
121        // TODO: Remove explitate vtable set once trac#186 is fixed.
122        ThreadCancelled(T) except;
123        except.virtual_table = &_default_vtable;
124        except.the_thread = &thrd;
125        except.the_exception = __cfaehm_cancellation_exception( cancellation );
126        // Why is this cast required?
127        throwResume (ThreadCancelled(T) &)except;
128
129        except.the_exception->virtual_table->free( except.the_exception );
130        free( cancellation );
131        desc->self_cor.cancellation = 0p;
132}
133
134void ^?{}( thread_dtor_guard_t & this ) {
135        ^(this.mg){};
136}
137
138//-----------------------------------------------------------------------------
139// Starting and stopping threads
140forall( T & | is_thread(T) )
141void __thrd_start( T & this, void (*main_p)(T &) ) {
142        thread$ * this_thrd = get_thread(this);
143
144        disable_interrupts();
145        __cfactx_start(main_p, get_coroutine(this), this, __cfactx_invoke_thread);
146
147        this_thrd->context.[SP, FP] = this_thrd->self_cor.context.[SP, FP];
148        /* paranoid */ verify( this_thrd->context.SP );
149
150        schedule_thread$( this_thrd, UNPARK_LOCAL );
151        enable_interrupts();
152}
153
154//-----------------------------------------------------------------------------
155// Support for threads that don't ues the thread keyword
156forall( T & | sized(T) | is_thread(T) | { void ?{}(T&); } )
157void ?{}( scoped(T)& this ) with( this ) {
158        handle{};
159        __thrd_start(handle, main);
160}
161
162forall( T &, P... | sized(T) | is_thread(T) | { void ?{}(T&, P); } )
163void ?{}( scoped(T)& this, P params ) with( this ) {
164        handle{ params };
165        __thrd_start(handle, main);
166}
167
168forall( T & | sized(T) | is_thread(T) )
169void ^?{}( scoped(T)& this ) with( this ) {
170        ^handle{};
171}
172
173//-----------------------------------------------------------------------------
174forall(T & | is_thread(T) | IS_RESUMPTION_EXCEPTION(ThreadCancelled(T))
175        | { EHM_DEFAULT_VTABLE(ThreadCancelled(T)); })
176T & join( T & this ) {
177        thread_dtor_guard_t guard = { this, defaultResumptionHandler };
178        return this;
179}
180
181//-----------------------------------------------------------------------------
182bool migrate( thread$ * thrd, struct cluster & cl ) {
183        monitor$ * tmon = get_monitor(thrd);
184        monitor$ * __monitors[] = { tmon };
185        monitor_guard_t __guard = { __monitors, 1 };
186        {
187                // if nothing needs to be done, return false
188                if( thrd->curr_cluster == &cl ) return false;
189
190                // are we migrating ourself?
191                const bool local = thrd == active_thread();
192
193                /* paranoid */ verify( !local || &cl != active_cluster() );
194                /* paranoid */ verify( !local || thrd->curr_cluster == active_cluster() );
195                /* paranoid */ verify( !local || thrd->curr_cluster == active_processor()->cltr );
196                /* paranoid */ verify( local || tmon->signal_stack.top->owner->waiting_thread == thrd );
197                /* paranoid */ verify( local || tmon->signal_stack.top );
198
199                // make sure we aren't interrupted while doing this
200                // not as important if we aren't local
201                disable_interrupts();
202
203                // actually move the thread
204                unregister( thrd->curr_cluster, *thrd );
205                thrd->curr_cluster = &cl;
206                doregister( thrd->curr_cluster, *thrd );
207
208                // restore interrupts
209                enable_interrupts();
210
211                // if this is the local thread, we are still running on the old cluster
212                if(local) yield();
213
214                /* paranoid */ verify( !local || &cl == active_cluster() );
215                /* paranoid */ verify( !local || thrd->curr_cluster == active_cluster() );
216                /* paranoid */ verify( !local || thrd->curr_cluster == active_processor()->cltr );
217                /* paranoid */ verify(  local || tmon->signal_stack.top );
218                /* paranoid */ verify(  local || tmon->signal_stack.top->owner->waiting_thread == thrd );
219
220                return true;
221        }
222}
223
224//-----------------------------------------------------------------------------
225
226void set_seed( size_t seed ) {
227        PRNG_STATE_T & state = active_thread()->random_state;
228        PRNG_SET_SEED( state, seed );
229        __global_random_seed = seed;
230        __global_random_prime = seed;
231        __global_random_mask = true;
232} // set_seed
233
234size_t prng( void ) {                                                                   // [0,UINT_MAX]
235        return PRNG_NAME( active_thread()->random_state );
236} // prng
237
238// Local Variables: //
239// mode: c //
240// tab-width: 4 //
241// End: //
Note: See TracBrowser for help on using the repository browser.