kamkow1/my-os-project2
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Handle continous jobs via new subsystem - CJob

Browse Source
This commit is contained in:
Kamil Kowalczyk
2025-11-24 21:53:51 +01:00
parent 4f55d765b4
commit 80a29d8ff6
7 changed files with 130 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
kernel/ipc/mbus/mbus.c
View File

@ -7,15 +7,47 @@
#include "std/string.h"
#include "dlmalloc/malloc.h"
#include "util/util.h"
#include "cjob/cjob.h"
#include "hshtb.h"
#include "errors.h"
#include "kprintf.h"
IpcMBuses IPC_MBUSES;
void ipc_mbus_gc_cjob(void *arg) {
for (size_t i = 0; i < LEN(IPC_MBUSES.mbuses); i++) {
spinlock_acquire(&IPC_MBUSES.spinlock);
IpcMBus *mbus = &IPC_MBUSES.mbuses[i];
spinlock_release(&IPC_MBUSES.spinlock);
spinlock_acquire(&mbus->spinlock);
if (mbus->_hshtbstate != HSHTB_TAKEN) {
spinlock_release(&mbus->spinlock);
continue;
}
IpcMBusCons *cons, *constmp;
LL_FOREACH_SAFE(mbus->consumers, cons, constmp) {
spinlock_acquire(&PROCS.spinlock);
Proc *proc = NULL;
LL_FINDPROP(PROCS.procs, proc, pid, cons->pid);
spinlock_release(&PROCS.spinlock);
if (proc == NULL) {
LL_REMOVE(mbus->consumers, cons);
dlfree(cons->rbuf.buffer);
dlfree(cons);
}
}
spinlock_release(&mbus->spinlock);
}
}
void ipc_mbusinit(void) {
memset(&IPC_MBUSES, 0, sizeof(IPC_MBUSES));
spinlock_init(&IPC_MBUSES.spinlock);
cjob_register(&ipc_mbus_gc_cjob, NULL);
}
IpcMBus *ipc_mbusmake(const char *name, size_t objsize, size_t objmax) {
@ -174,32 +206,3 @@ int32_t ipc_mbusdttch(const char *name, uint64_t pid) {
return E_OK;
}
void ipc_mbustick(void) {
for (size_t i = 0; i < LEN(IPC_MBUSES.mbuses); i++) {
spinlock_acquire(&IPC_MBUSES.spinlock);
IpcMBus *mbus = &IPC_MBUSES.mbuses[i];
spinlock_release(&IPC_MBUSES.spinlock);
spinlock_acquire(&mbus->spinlock);
if (mbus->_hshtbstate != HSHTB_TAKEN) {
spinlock_release(&mbus->spinlock);
continue;
}
IpcMBusCons *cons, *constmp;
LL_FOREACH_SAFE(mbus->consumers, cons, constmp) {
spinlock_acquire(&PROCS.spinlock);
Proc *proc = NULL;
LL_FINDPROP(PROCS.procs, proc, pid, cons->pid);
spinlock_release(&PROCS.spinlock);
if (proc == NULL) {
LL_REMOVE(mbus->consumers, cons);
dlfree(cons->rbuf.buffer);
dlfree(cons);
}
}
spinlock_release(&mbus->spinlock);
}
}

1
kernel/ipc/mbus/mbus.h
View File

@ -34,6 +34,5 @@ int32_t ipc_mbuspublish(const char *name, const uint8_t *const buffer);
int32_t ipc_mbusconsume(const char *name, uint8_t *const buffer, uint64_t pid);
int32_t ipc_mbusattch(const char *name, uint64_t pid);
int32_t ipc_mbusdttch(const char *name, uint64_t pid);
void ipc_mbustick(void);
#endif // IPC_MBUS_MBUS_H_