6033849287

Committed 31 Aug 2023 06:28AM UTC coverage: 95.806% (-0.3%) from 96.077%

Build Type

push

github

Committed by Chris Dragan

Commit Message

net: add blocking property (read-only)

Run Details

21 of 21 new or added lines in 1 file covered. (100.0%)

23827 of 24870 relevant lines covered (95.81%)

1070731.17 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.89

/modules/kos_mod_threads.c

/* SPDX-License-Identifier: MIT
 * Copyright (c) 2014-2023 Chris Dragan
 */

#include "../inc/kos_array.h"
#include "../inc/kos_constants.h"
#include "../inc/kos_error.h"
#include "../inc/kos_instance.h"
#include "../inc/kos_malloc.h"
#include "../inc/kos_module.h"
#include "../inc/kos_object.h"
#include "../inc/kos_threads.h"
#include "../inc/kos_utils.h"
#include "../core/kos_debug.h"
#include "../core/kos_misc.h"
#include "../core/kos_try.h"

#define KOS_MAX_SEM 0x7FFFFFFF

KOS_DECLARE_STATIC_CONST_STRING(str_count,               "count");
KOS_DECLARE_STATIC_CONST_STRING(str_err_bad_module,      "failed to get private data from module thread");
KOS_DECLARE_STATIC_CONST_STRING(str_err_cond_var_failed, "failed to create a condition variable");
KOS_DECLARE_STATIC_CONST_STRING(str_err_count_too_small, "count argument is less than 1");
KOS_DECLARE_STATIC_CONST_STRING(str_err_count_too_large, "count argument exceeds 0x7FFFFFFF");
KOS_DECLARE_STATIC_CONST_STRING(str_err_mutex_failed,    "failed to create a mutex");
KOS_DECLARE_STATIC_CONST_STRING(str_err_init_too_large,  "init argument exceeds 0x7FFFFFFF");
KOS_DECLARE_STATIC_CONST_STRING(str_err_init_too_small,  "init argument is less than 0");
KOS_DECLARE_STATIC_CONST_STRING(str_init,                "init");

KOS_DECLARE_PRIVATE_CLASS(mutex_priv_class);

static void mutex_finalize(KOS_CONTEXT ctx,
                           void       *priv)
{
    if (priv)
        kos_destroy_mutex((KOS_MUTEX *)&priv);
}

/* @item threads mutex()
 *
 *     mutex()
 *
 * Mutex object class.
 *
 * Mutex objects are best used with the `with` statement.
 */
static KOS_OBJ_ID mutex_ctor(KOS_CONTEXT ctx,
                             KOS_OBJ_ID  this_obj,
                             KOS_OBJ_ID  args_obj)
{
    KOS_LOCAL  mutex;
    KOS_OBJ_ID proto;
    KOS_MUTEX  mutex_obj = KOS_NULL;
    int        error     = KOS_SUCCESS;

    KOS_init_local(ctx, &mutex);

    error = kos_create_mutex(&mutex_obj);

    if (error)
        RAISE_EXCEPTION_STR(str_err_mutex_failed);

    proto = KOS_get_module(ctx);
    TRY_OBJID(proto);

    proto = KOS_atomic_read_relaxed_obj(OBJPTR(MODULE, proto)->priv);
    if (IS_BAD_PTR(proto) || kos_seq_fail())
        RAISE_EXCEPTION_STR(str_err_bad_module);

    proto = KOS_array_read(ctx, proto, 0);
    TRY_OBJID(proto);

    mutex.o = KOS_new_object_with_private(ctx,
                                          proto,
                                          &mutex_priv_class,
                                          mutex_finalize);
    TRY_OBJID(mutex.o);

    KOS_object_set_private_ptr(mutex.o, mutex_obj);
    mutex_obj = KOS_NULL;

cleanup:
    if (mutex_obj)
        kos_destroy_mutex(&mutex_obj);

    mutex.o = KOS_destroy_top_local(ctx, &mutex);

    return error ? KOS_BADPTR : mutex.o;
}

/* @item threads mutex.prototype.acquire()
 *
 *     mutex.prototype.acquire()
 *
 * Locks the mutex object.
 *
 * If the mutex is already locked by another thread, this function will wait
 * until it is unlocked.
 *
 * Returns `this` mutex object.
 */
static KOS_OBJ_ID mutex_acquire(KOS_CONTEXT ctx,
                                KOS_OBJ_ID  this_obj,
                                KOS_OBJ_ID  args_obj)
{
    KOS_LOCAL mutex;
    KOS_MUTEX mutex_obj;

    KOS_init_local_with(ctx, &mutex, this_obj);

    mutex_obj = (KOS_MUTEX)KOS_object_get_private(mutex.o, &mutex_priv_class);

    if (mutex_obj) {
        KOS_suspend_context(ctx);

        kos_lock_mutex(mutex_obj);

        KOS_resume_context(ctx);
    }

    return KOS_destroy_top_local(ctx, &mutex);
}

/* @item threads mutex.prototype.release()
 *
 *     mutex.prototype.release()
 *
 * Unlocks the mutex object, if it is held by the current thread.
 *
 * If the mutex is not held by the current thread, this function does nothing.
 *
 * Returns `this` mutex object.
 */
static KOS_OBJ_ID mutex_release(KOS_CONTEXT ctx,
                                KOS_OBJ_ID  this_obj,
                                KOS_OBJ_ID  args_obj)
{
    const KOS_MUTEX mutex_obj = (KOS_MUTEX)KOS_object_get_private(this_obj, &mutex_priv_class);

    if (mutex_obj)
        kos_unlock_mutex(mutex_obj);

    return this_obj;
}

typedef struct KOS_SEMAPHORE_S {
    KOS_MUTEX            mutex;
    KOS_COND_VAR         cond_var;
    KOS_ATOMIC(uint32_t) value;
} KOS_SEMAPHORE;

KOS_DECLARE_PRIVATE_CLASS(semaphore_priv_class);

static void semaphore_finalize(KOS_CONTEXT ctx,
                               void       *priv)
{
    if (priv) {
        KOS_SEMAPHORE *const sem = (KOS_SEMAPHORE *)priv;

        if (sem->mutex)
            kos_destroy_mutex(&sem->mutex);

        if (sem->cond_var)
            kos_destroy_cond_var(&sem->cond_var);

        KOS_free(sem);
    }
}

static const KOS_CONVERT sem_args[2] = {
    KOS_DEFINE_OPTIONAL_ARG(str_init, TO_SMALL_INT(0)),
    KOS_DEFINE_TAIL_ARG()
};

/* @item threads semaphore()
 *
 *     semaphore(init = 0)
 *
 * Semaphore object class.
 *
 * A semaphore is an integer number which can be incremented (release)
 * or decremented (acquire).  If an `acquire()` function is called on
 * a semaphore which has a zero-value, the function will block until
 * another thread increments the semaphore.
 *
 * `init` is the initial integer value for the new semaphore object.
 *
 * Semaphore objects can be used with the `with` statement.
 */
static KOS_OBJ_ID semaphore_ctor(KOS_CONTEXT ctx,
                                 KOS_OBJ_ID  this_obj,
                                 KOS_OBJ_ID  args_obj)
{
    KOS_LOCAL      semaphore;
    KOS_OBJ_ID     proto;
    KOS_SEMAPHORE *sem   = KOS_NULL;
    int            error = KOS_SUCCESS;

    assert(KOS_get_array_size(args_obj) >= 1);

    KOS_init_local(ctx, &semaphore);

    sem = (KOS_SEMAPHORE *)KOS_malloc(sizeof(KOS_SEMAPHORE));

    if ( ! sem) {
        KOS_raise_exception(ctx, KOS_STR_OUT_OF_MEMORY);
        goto cleanup;
    }

    sem->mutex    = KOS_NULL;
    sem->cond_var = KOS_NULL;

    {
        int64_t value64;

        KOS_OBJ_ID arg = KOS_array_read(ctx, args_obj, 0);
        TRY_OBJID(arg);

        TRY(KOS_get_integer(ctx, arg, &value64));

        if (value64 < 0)
            RAISE_EXCEPTION_STR(str_err_init_too_small);

        if (value64 > KOS_MAX_SEM)
            RAISE_EXCEPTION_STR(str_err_init_too_large);

        KOS_atomic_write_relaxed_u32(sem->value, (uint32_t)value64);
    }

    error = kos_create_mutex(&sem->mutex);

    if (error)
        RAISE_EXCEPTION_STR(str_err_mutex_failed);

    error = kos_create_cond_var(&sem->cond_var);

    if (error)
        RAISE_EXCEPTION_STR(str_err_cond_var_failed);

    proto = KOS_get_module(ctx);
    TRY_OBJID(proto);

    proto = KOS_atomic_read_relaxed_obj(OBJPTR(MODULE, proto)->priv);
    if (IS_BAD_PTR(proto) || kos_seq_fail())
        RAISE_EXCEPTION_STR(str_err_bad_module);

    proto = KOS_array_read(ctx, proto, 1);
    TRY_OBJID(proto);

    semaphore.o = KOS_new_object_with_private(ctx,
                                              proto,
                                              &semaphore_priv_class,
                                              semaphore_finalize);
    TRY_OBJID(semaphore.o);

    KOS_object_set_private_ptr(semaphore.o, sem);
    sem = KOS_NULL;

cleanup:
    if (sem)
        semaphore_finalize(ctx, sem);

    semaphore.o = KOS_destroy_top_local(ctx, &semaphore);

    return error ? KOS_BADPTR : semaphore.o;
}

static const KOS_CONVERT count_arg[2] = {
    { KOS_CONST_ID(str_count), TO_SMALL_INT(1), 0, 0, KOS_NATIVE_INT64 },
    KOS_DEFINE_TAIL_ARG()
};

static int get_count_arg(KOS_CONTEXT ctx,
                         KOS_OBJ_ID  args_obj,
                         uint32_t   *count)
{
    int64_t count64 = 0;
    int     error;

    error = KOS_extract_native_from_array(ctx, args_obj, "argument", count_arg, KOS_NULL, &count64);
    if (error)
        return error;

    if (count64 < 1) {
        KOS_raise_exception(ctx, KOS_CONST_ID(str_err_count_too_small));
        return KOS_ERROR_EXCEPTION;
    }

    if (count64 > KOS_MAX_SEM) {
        KOS_raise_exception(ctx, KOS_CONST_ID(str_err_count_too_large));
        return KOS_ERROR_EXCEPTION;
    }

    *count = (uint32_t)(uint64_t)count64;

    return KOS_SUCCESS;
}

/* @item threads semaphore.prototype.acquire()
 *
 *     semaphore.prototype.acquire(count = 1)
 *
 * Subtracts `count` from the semaphore value.
 *
 * `count` defaults to 1.  If `count` is less than 1 or greater than 0x7FFFFFFF
 * throws an exception.
 *
 * If the semaphore value is already 0, blocks until another thread increments it,
 * then performs the decrement operation.  This is repeated until the value has
 * been decremented `count` times.  The decrement operation is non-atomic meaning
 * that if two threads are trying to acquire with `count > 1`, each of them could
 * decrement the value by 1 multiple times.
 *
 * Returns `this` semaphore object.
 */
static KOS_OBJ_ID semaphore_acquire(KOS_CONTEXT ctx,
                                    KOS_OBJ_ID  this_obj,
                                    KOS_OBJ_ID  args_obj)
{
    KOS_LOCAL      semaphore;
    KOS_SEMAPHORE *sem   = KOS_NULL;
    uint32_t       count = 0;

    KOS_init_local_with(ctx, &semaphore, this_obj);

    sem = (KOS_SEMAPHORE *)KOS_object_get_private(semaphore.o, &semaphore_priv_class);

    if (get_count_arg(ctx, args_obj, &count)) {
        KOS_destroy_top_local(ctx, &semaphore);
        return KOS_BADPTR;
    }

    if (sem) {

        int suspended = 0;

        do {
            const uint32_t old_value = KOS_atomic_read_relaxed_u32(sem->value);
            const uint32_t dec_value = (old_value > count) ? count : old_value;

            if ( ! dec_value) {

                if ( ! suspended) {
                    KOS_suspend_context(ctx);

                    kos_lock_mutex(sem->mutex);

                    suspended = 1;

                    continue;
                }

                kos_wait_cond_var(sem->cond_var, sem->mutex);

                continue;
            }

            if ( ! KOS_atomic_cas_weak_u32(sem->value, old_value, old_value - dec_value))
                continue;

            count -= dec_value;
        } while (count);

        if (suspended) {
            kos_unlock_mutex(sem->mutex);

            KOS_resume_context(ctx);
        }
    }

    return KOS_destroy_top_local(ctx, &semaphore);
}

/* @item threads semaphore.prototype.release()
 *
 *     semaphore.prototype.release()
 *
 * Increments the semaphore value and signals other threads that may be
 * waiting on `acquire()`.
 *
 * Returns `this` semaphore object.
 */
static KOS_OBJ_ID semaphore_release(KOS_CONTEXT ctx,
                                    KOS_OBJ_ID  this_obj,
                                    KOS_OBJ_ID  args_obj)
{
    KOS_LOCAL      semaphore;
    KOS_SEMAPHORE *sem   = KOS_NULL;
    uint32_t       count = 0;

    KOS_init_local_with(ctx, &semaphore, this_obj);

    sem = (KOS_SEMAPHORE *)KOS_object_get_private(semaphore.o, &semaphore_priv_class);

    if (get_count_arg(ctx, args_obj, &count)) {
        KOS_destroy_top_local(ctx, &semaphore);
        return KOS_BADPTR;
    }

    if (sem) {
        uint32_t old_value;
        uint32_t max_inc;

        KOS_suspend_context(ctx);

        kos_lock_mutex(sem->mutex);

        do {
            old_value = KOS_atomic_read_relaxed_u32(sem->value);
            max_inc   = (uint32_t)KOS_MAX_SEM - old_value;

            if (count > max_inc) {
                kos_unlock_mutex(sem->mutex);

                KOS_resume_context(ctx);

                KOS_destroy_top_local(ctx, &semaphore);

                KOS_raise_printf(ctx, "semaphore value %u cannot be increased by %u",
                                 old_value, count);
                return KOS_BADPTR;
            }

        } while ( ! KOS_atomic_cas_weak_u32(sem->value, old_value, old_value + count));

        kos_unlock_mutex(sem->mutex);

        kos_broadcast_cond_var(sem->cond_var);

        KOS_resume_context(ctx);
    }

    return KOS_destroy_top_local(ctx, &semaphore);
}

static KOS_OBJ_ID semaphore_value(KOS_CONTEXT ctx,
                                  KOS_OBJ_ID  this_obj,
                                  KOS_OBJ_ID  args_obj)
{
    KOS_SEMAPHORE *const sem = (KOS_SEMAPHORE *)KOS_object_get_private(this_obj, &semaphore_priv_class);

    if (sem) {
        const uint32_t value = KOS_atomic_read_relaxed_u32(sem->value);

        return KOS_new_int(ctx, (int64_t)value);
    }

    return KOS_VOID;
}

int kos_module_threads_init(KOS_CONTEXT ctx, KOS_OBJ_ID module_obj)
{
    int       error = KOS_SUCCESS;
    KOS_LOCAL module;
    KOS_LOCAL priv;
    KOS_LOCAL mutex_proto;
    KOS_LOCAL semaphore_proto;

    KOS_init_local_with(ctx, &module, module_obj);
    KOS_init_locals(    ctx, &priv, &mutex_proto, &semaphore_proto, kos_end_locals);

    priv.o = KOS_new_array(ctx, 2);
    TRY_OBJID(priv.o);

    KOS_atomic_write_relaxed_ptr(OBJPTR(MODULE, module.o)->priv, priv.o);

    TRY_ADD_CONSTRUCTOR(    ctx, module.o,                    "mutex",     mutex_ctor,        KOS_NULL, &mutex_proto.o);
    TRY_ADD_MEMBER_FUNCTION(ctx, module.o, mutex_proto.o,     "acquire",   mutex_acquire,     KOS_NULL);
    TRY_ADD_MEMBER_FUNCTION(ctx, module.o, mutex_proto.o,     "release",   mutex_release,     KOS_NULL);

    TRY_ADD_CONSTRUCTOR(    ctx, module.o,                    "semaphore", semaphore_ctor,    sem_args, &semaphore_proto.o);
    TRY_ADD_MEMBER_FUNCTION(ctx, module.o, semaphore_proto.o, "acquire",   semaphore_acquire, count_arg);
    TRY_ADD_MEMBER_FUNCTION(ctx, module.o, semaphore_proto.o, "release",   semaphore_release, count_arg);
    TRY_ADD_MEMBER_PROPERTY(ctx, module.o, semaphore_proto.o, "value",     semaphore_value,   KOS_NULL);

    TRY(KOS_array_write(ctx, priv.o, 0, mutex_proto.o));
    TRY(KOS_array_write(ctx, priv.o, 1, semaphore_proto.o));

cleanup:
    KOS_destroy_top_locals(ctx, &priv, &module);

    return error;
}

1	/* SPDX-License-Identifier: MIT
2	* Copyright (c) 2014-2023 Chris Dragan
3	*/
4
5	#include "../inc/kos_array.h"
6	#include "../inc/kos_constants.h"
7	#include "../inc/kos_error.h"
8	#include "../inc/kos_instance.h"
9	#include "../inc/kos_malloc.h"
10	#include "../inc/kos_module.h"
11	#include "../inc/kos_object.h"
12	#include "../inc/kos_threads.h"
13	#include "../inc/kos_utils.h"
14	#include "../core/kos_debug.h"
15	#include "../core/kos_misc.h"
16	#include "../core/kos_try.h"
17
18	#define KOS_MAX_SEM 0x7FFFFFFF
19
20	KOS_DECLARE_STATIC_CONST_STRING(str_count, "count");
21	KOS_DECLARE_STATIC_CONST_STRING(str_err_bad_module, "failed to get private data from module thread");
22	KOS_DECLARE_STATIC_CONST_STRING(str_err_cond_var_failed, "failed to create a condition variable");
23	KOS_DECLARE_STATIC_CONST_STRING(str_err_count_too_small, "count argument is less than 1");
24	KOS_DECLARE_STATIC_CONST_STRING(str_err_count_too_large, "count argument exceeds 0x7FFFFFFF");
25	KOS_DECLARE_STATIC_CONST_STRING(str_err_mutex_failed, "failed to create a mutex");
26	KOS_DECLARE_STATIC_CONST_STRING(str_err_init_too_large, "init argument exceeds 0x7FFFFFFF");
27	KOS_DECLARE_STATIC_CONST_STRING(str_err_init_too_small, "init argument is less than 0");
28	KOS_DECLARE_STATIC_CONST_STRING(str_init, "init");
29
30	KOS_DECLARE_PRIVATE_CLASS(mutex_priv_class);
31
32	static void mutex_finalize(KOS_CONTEXT ctx,	6✔
33	void *priv)
34	{
35	if (priv)	6✔
36	kos_destroy_mutex((KOS_MUTEX *)&priv);	6✔
37	}	6✔
38
39	/* @item threads mutex()
40	*
41	* mutex()
42	*
43	* Mutex object class.
44	*
45	* Mutex objects are best used with the `with` statement.
46	*/
47	static KOS_OBJ_ID mutex_ctor(KOS_CONTEXT ctx,	6✔
48	KOS_OBJ_ID this_obj,
49	KOS_OBJ_ID args_obj)
50	{
51	KOS_LOCAL mutex;
52	KOS_OBJ_ID proto;
53	KOS_MUTEX mutex_obj = KOS_NULL;	6✔
54	int error = KOS_SUCCESS;	6✔
55
56	KOS_init_local(ctx, &mutex);	6✔
57
58	error = kos_create_mutex(&mutex_obj);	6✔
59
60	if (error)	6✔
61	RAISE_EXCEPTION_STR(str_err_mutex_failed);	×
62
63	proto = KOS_get_module(ctx);	6✔
64	TRY_OBJID(proto);	6✔
65
66	proto = KOS_atomic_read_relaxed_obj(OBJPTR(MODULE, proto)->priv);	6✔
67	if (IS_BAD_PTR(proto) \|\| kos_seq_fail())	6✔
68	RAISE_EXCEPTION_STR(str_err_bad_module);	×
69
70	proto = KOS_array_read(ctx, proto, 0);	6✔
71	TRY_OBJID(proto);	6✔
72
73	mutex.o = KOS_new_object_with_private(ctx,	6✔
74	proto,
75	&mutex_priv_class,
76	mutex_finalize);
77	TRY_OBJID(mutex.o);	6✔
78
79	KOS_object_set_private_ptr(mutex.o, mutex_obj);	6✔
80	mutex_obj = KOS_NULL;	6✔
81
82	cleanup:	6✔
83	if (mutex_obj)	6✔
84	kos_destroy_mutex(&mutex_obj);	×
85
86	mutex.o = KOS_destroy_top_local(ctx, &mutex);	6✔
87
88	return error ? KOS_BADPTR : mutex.o;	6✔
89	}
90
91	/* @item threads mutex.prototype.acquire()
92	*
93	* mutex.prototype.acquire()
94	*
95	* Locks the mutex object.
96	*
97	* If the mutex is already locked by another thread, this function will wait
98	* until it is unlocked.
99	*
100	* Returns `this` mutex object.
101	*/
102	static KOS_OBJ_ID mutex_acquire(KOS_CONTEXT ctx,	43✔
103	KOS_OBJ_ID this_obj,
104	KOS_OBJ_ID args_obj)
105	{
106	KOS_LOCAL mutex;
107	KOS_MUTEX mutex_obj;
108
109	KOS_init_local_with(ctx, &mutex, this_obj);	43✔
110
111	mutex_obj = (KOS_MUTEX)KOS_object_get_private(mutex.o, &mutex_priv_class);	43✔
112
113	if (mutex_obj) {	43✔
114	KOS_suspend_context(ctx);	43✔
115
116	kos_lock_mutex(mutex_obj);	43✔
117
118	KOS_resume_context(ctx);	43✔
119	}
120
121	return KOS_destroy_top_local(ctx, &mutex);	43✔
122	}
123
124	/* @item threads mutex.prototype.release()
125	*
126	* mutex.prototype.release()
127	*
128	* Unlocks the mutex object, if it is held by the current thread.
129	*
130	* If the mutex is not held by the current thread, this function does nothing.
131	*
132	* Returns `this` mutex object.
133	*/
134	static KOS_OBJ_ID mutex_release(KOS_CONTEXT ctx,	43✔
135	KOS_OBJ_ID this_obj,
136	KOS_OBJ_ID args_obj)
137	{
138	const KOS_MUTEX mutex_obj = (KOS_MUTEX)KOS_object_get_private(this_obj, &mutex_priv_class);	43✔
139
140	if (mutex_obj)	43✔
141	kos_unlock_mutex(mutex_obj);	43✔
142
143	return this_obj;	43✔
144	}
145
146	typedef struct KOS_SEMAPHORE_S {
147	KOS_MUTEX mutex;
148	KOS_COND_VAR cond_var;
149	KOS_ATOMIC(uint32_t) value;
150	} KOS_SEMAPHORE;
151
152	KOS_DECLARE_PRIVATE_CLASS(semaphore_priv_class);
153
154	static void semaphore_finalize(KOS_CONTEXT ctx,	8✔
155	void *priv)
156	{
157	if (priv) {	8✔
158	KOS_SEMAPHORE const sem = (KOS_SEMAPHORE )priv;	8✔
159
160	if (sem->mutex)	8✔
161	kos_destroy_mutex(&sem->mutex);	6✔
162
163	if (sem->cond_var)	8✔
164	kos_destroy_cond_var(&sem->cond_var);	6✔
165
166	KOS_free(sem);	8✔
167	}
168	}	8✔
169
170	static const KOS_CONVERT sem_args[2] = {
171	KOS_DEFINE_OPTIONAL_ARG(str_init, TO_SMALL_INT(0)),
172	KOS_DEFINE_TAIL_ARG()
173	};
174
175	/* @item threads semaphore()
176	*
177	* semaphore(init = 0)
178	*
179	* Semaphore object class.
180	*
181	* A semaphore is an integer number which can be incremented (release)
182	* or decremented (acquire). If an `acquire()` function is called on
183	* a semaphore which has a zero-value, the function will block until
184	* another thread increments the semaphore.
185	*
186	* `init` is the initial integer value for the new semaphore object.
187	*
188	* Semaphore objects can be used with the `with` statement.
189	*/
190	static KOS_OBJ_ID semaphore_ctor(KOS_CONTEXT ctx,	8✔
191	KOS_OBJ_ID this_obj,
192	KOS_OBJ_ID args_obj)
193	{
194	KOS_LOCAL semaphore;
195	KOS_OBJ_ID proto;
196	KOS_SEMAPHORE *sem = KOS_NULL;	8✔
197	int error = KOS_SUCCESS;	8✔
198
199	assert(KOS_get_array_size(args_obj) >= 1);	8✔
200
201	KOS_init_local(ctx, &semaphore);	8✔
202
203	sem = (KOS_SEMAPHORE *)KOS_malloc(sizeof(KOS_SEMAPHORE));	8✔
204
205	if ( ! sem) {	8✔
206	KOS_raise_exception(ctx, KOS_STR_OUT_OF_MEMORY);	×
207	goto cleanup;	×
208	}
209
210	sem->mutex = KOS_NULL;	8✔
211	sem->cond_var = KOS_NULL;	8✔
212
213	{
214	int64_t value64;
215
216	KOS_OBJ_ID arg = KOS_array_read(ctx, args_obj, 0);	8✔
217	TRY_OBJID(arg);	10✔
218
219	TRY(KOS_get_integer(ctx, arg, &value64));	8✔
220
221	if (value64 < 0)	8✔
222	RAISE_EXCEPTION_STR(str_err_init_too_small);	1✔
223
224	if (value64 > KOS_MAX_SEM)	7✔
225	RAISE_EXCEPTION_STR(str_err_init_too_large);	1✔
226
227	KOS_atomic_write_relaxed_u32(sem->value, (uint32_t)value64);	6✔
228	}
229
230	error = kos_create_mutex(&sem->mutex);	6✔
231
232	if (error)	6✔
233	RAISE_EXCEPTION_STR(str_err_mutex_failed);	×
234
235	error = kos_create_cond_var(&sem->cond_var);	6✔
236
237	if (error)	6✔
238	RAISE_EXCEPTION_STR(str_err_cond_var_failed);	×
239
240	proto = KOS_get_module(ctx);	6✔
241	TRY_OBJID(proto);	6✔
242
243	proto = KOS_atomic_read_relaxed_obj(OBJPTR(MODULE, proto)->priv);	6✔
244	if (IS_BAD_PTR(proto) \|\| kos_seq_fail())	6✔
245	RAISE_EXCEPTION_STR(str_err_bad_module);	×
246
247	proto = KOS_array_read(ctx, proto, 1);	6✔
248	TRY_OBJID(proto);	6✔
249
250	semaphore.o = KOS_new_object_with_private(ctx,	6✔
251	proto,
252	&semaphore_priv_class,
253	semaphore_finalize);
254	TRY_OBJID(semaphore.o);	6✔
255
256	KOS_object_set_private_ptr(semaphore.o, sem);	6✔
257	sem = KOS_NULL;	6✔
258
259	cleanup:	8✔
260	if (sem)	8✔
261	semaphore_finalize(ctx, sem);	2✔
262
263	semaphore.o = KOS_destroy_top_local(ctx, &semaphore);	8✔
264
265	return error ? KOS_BADPTR : semaphore.o;	8✔
266	}
267
268	static const KOS_CONVERT count_arg[2] = {
269	{ KOS_CONST_ID(str_count), TO_SMALL_INT(1), 0, 0, KOS_NATIVE_INT64 },
270	KOS_DEFINE_TAIL_ARG()
271	};
272
273	static int get_count_arg(KOS_CONTEXT ctx,	49✔
274	KOS_OBJ_ID args_obj,
275	uint32_t *count)
276	{
277	int64_t count64 = 0;	49✔
278	int error;
279
280	error = KOS_extract_native_from_array(ctx, args_obj, "argument", count_arg, KOS_NULL, &count64);	49✔
281	if (error)	49✔
282	return error;	×
283
284	if (count64 < 1) {	49✔
285	KOS_raise_exception(ctx, KOS_CONST_ID(str_err_count_too_small));	2✔
286	return KOS_ERROR_EXCEPTION;	2✔
287	}
288
289	if (count64 > KOS_MAX_SEM) {	47✔
290	KOS_raise_exception(ctx, KOS_CONST_ID(str_err_count_too_large));	2✔
291	return KOS_ERROR_EXCEPTION;	2✔
292	}
293
294	*count = (uint32_t)(uint64_t)count64;	45✔
295
296	return KOS_SUCCESS;	45✔
297	}
298
299	/* @item threads semaphore.prototype.acquire()
300	*
301	* semaphore.prototype.acquire(count = 1)
302	*
303	* Subtracts `count` from the semaphore value.
304	*
305	* `count` defaults to 1. If `count` is less than 1 or greater than 0x7FFFFFFF
306	* throws an exception.
307	*
308	* If the semaphore value is already 0, blocks until another thread increments it,
309	* then performs the decrement operation. This is repeated until the value has
310	* been decremented `count` times. The decrement operation is non-atomic meaning
311	* that if two threads are trying to acquire with `count > 1`, each of them could
312	* decrement the value by 1 multiple times.
313	*
314	* Returns `this` semaphore object.
315	*/
316	static KOS_OBJ_ID semaphore_acquire(KOS_CONTEXT ctx,	33✔
317	KOS_OBJ_ID this_obj,
318	KOS_OBJ_ID args_obj)
319	{
320	KOS_LOCAL semaphore;
321	KOS_SEMAPHORE *sem = KOS_NULL;	33✔
322	uint32_t count = 0;	33✔
323
324	KOS_init_local_with(ctx, &semaphore, this_obj);	33✔
325
326	sem = (KOS_SEMAPHORE *)KOS_object_get_private(semaphore.o, &semaphore_priv_class);	33✔
327
328	if (get_count_arg(ctx, args_obj, &count)) {	33✔
329	KOS_destroy_top_local(ctx, &semaphore);	2✔
330	return KOS_BADPTR;	2✔
331	}
332
333	if (sem) {	31✔
334
335	int suspended = 0;	31✔
336
337	do {
338	const uint32_t old_value = KOS_atomic_read_relaxed_u32(sem->value);	31✔
339	const uint32_t dec_value = (old_value > count) ? count : old_value;	31✔
340
341	if ( ! dec_value) {	31✔
342
343	if ( ! suspended) {	×
344	KOS_suspend_context(ctx);	×
345
346	kos_lock_mutex(sem->mutex);	×
347
348	suspended = 1;	×
349
350	continue;	×
351	}
352
353	kos_wait_cond_var(sem->cond_var, sem->mutex);	×
354
355	continue;	×
356	}
357
358	if ( ! KOS_atomic_cas_weak_u32(sem->value, old_value, old_value - dec_value))	31✔
359	continue;	×
360
361	count -= dec_value;	31✔
362	} while (count);	31✔
363
364	if (suspended) {	31✔
365	kos_unlock_mutex(sem->mutex);	×
366
367	KOS_resume_context(ctx);	×
368	}
369	}
370
371	return KOS_destroy_top_local(ctx, &semaphore);	31✔
372	}
373
374	/* @item threads semaphore.prototype.release()
375	*
376	* semaphore.prototype.release()
377	*
378	* Increments the semaphore value and signals other threads that may be
379	* waiting on `acquire()`.
380	*
381	* Returns `this` semaphore object.
382	*/
383	static KOS_OBJ_ID semaphore_release(KOS_CONTEXT ctx,	16✔
384	KOS_OBJ_ID this_obj,
385	KOS_OBJ_ID args_obj)
386	{
387	KOS_LOCAL semaphore;
388	KOS_SEMAPHORE *sem = KOS_NULL;	16✔
389	uint32_t count = 0;	16✔
390
391	KOS_init_local_with(ctx, &semaphore, this_obj);	16✔
392
393	sem = (KOS_SEMAPHORE *)KOS_object_get_private(semaphore.o, &semaphore_priv_class);	16✔
394
395	if (get_count_arg(ctx, args_obj, &count)) {	16✔
396	KOS_destroy_top_local(ctx, &semaphore);	2✔
397	return KOS_BADPTR;	2✔
398	}
399
400	if (sem) {	14✔
401	uint32_t old_value;
402	uint32_t max_inc;
403
404	KOS_suspend_context(ctx);	14✔
405
406	kos_lock_mutex(sem->mutex);	14✔
407
408	do {
409	old_value = KOS_atomic_read_relaxed_u32(sem->value);	14✔
410	max_inc = (uint32_t)KOS_MAX_SEM - old_value;	14✔
411
412	if (count > max_inc) {	14✔
413	kos_unlock_mutex(sem->mutex);	1✔
414
415	KOS_resume_context(ctx);	1✔
416
417	KOS_destroy_top_local(ctx, &semaphore);	1✔
418
419	KOS_raise_printf(ctx, "semaphore value %u cannot be increased by %u",	1✔
420	old_value, count);
421	return KOS_BADPTR;	1✔
422	}
423
424	} while ( ! KOS_atomic_cas_weak_u32(sem->value, old_value, old_value + count));	13✔
425
426	kos_unlock_mutex(sem->mutex);	13✔
427
428	kos_broadcast_cond_var(sem->cond_var);	13✔
429
430	KOS_resume_context(ctx);	13✔
431	}
432
433	return KOS_destroy_top_local(ctx, &semaphore);	13✔
434	}
435
436	static KOS_OBJ_ID semaphore_value(KOS_CONTEXT ctx,	6✔
437	KOS_OBJ_ID this_obj,
438	KOS_OBJ_ID args_obj)
439	{
440	KOS_SEMAPHORE const sem = (KOS_SEMAPHORE )KOS_object_get_private(this_obj, &semaphore_priv_class);	6✔
441
442	if (sem) {	6✔
443	const uint32_t value = KOS_atomic_read_relaxed_u32(sem->value);	6✔
444
445	return KOS_new_int(ctx, (int64_t)value);	6✔
446	}
447
448	return KOS_VOID;	×
449	}
450
451	int kos_module_threads_init(KOS_CONTEXT ctx, KOS_OBJ_ID module_obj)	1✔
452	{
453	int error = KOS_SUCCESS;	1✔
454	KOS_LOCAL module;
455	KOS_LOCAL priv;
456	KOS_LOCAL mutex_proto;
457	KOS_LOCAL semaphore_proto;
458
459	KOS_init_local_with(ctx, &module, module_obj);	1✔
460	KOS_init_locals( ctx, &priv, &mutex_proto, &semaphore_proto, kos_end_locals);	1✔
461
462	priv.o = KOS_new_array(ctx, 2);	1✔
463	TRY_OBJID(priv.o);	1✔
464
465	KOS_atomic_write_relaxed_ptr(OBJPTR(MODULE, module.o)->priv, priv.o);	1✔
466
467	TRY_ADD_CONSTRUCTOR( ctx, module.o, "mutex", mutex_ctor, KOS_NULL, &mutex_proto.o);	1✔
468	TRY_ADD_MEMBER_FUNCTION(ctx, module.o, mutex_proto.o, "acquire", mutex_acquire, KOS_NULL);	1✔
469	TRY_ADD_MEMBER_FUNCTION(ctx, module.o, mutex_proto.o, "release", mutex_release, KOS_NULL);	1✔
470
471	TRY_ADD_CONSTRUCTOR( ctx, module.o, "semaphore", semaphore_ctor, sem_args, &semaphore_proto.o);	1✔
472	TRY_ADD_MEMBER_FUNCTION(ctx, module.o, semaphore_proto.o, "acquire", semaphore_acquire, count_arg);	1✔
473	TRY_ADD_MEMBER_FUNCTION(ctx, module.o, semaphore_proto.o, "release", semaphore_release, count_arg);	1✔
474	TRY_ADD_MEMBER_PROPERTY(ctx, module.o, semaphore_proto.o, "value", semaphore_value, KOS_NULL);	1✔
475
476	TRY(KOS_array_write(ctx, priv.o, 0, mutex_proto.o));	1✔
477	TRY(KOS_array_write(ctx, priv.o, 1, semaphore_proto.o));	1✔
478
479	cleanup:	1✔
480	KOS_destroy_top_locals(ctx, &priv, &module);	1✔
481
482	return error;	1✔
483	}

kos-lang / kos / 6033849287

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous