@@ -14,17 +14,18 @@ Syscall::Syscall(): _syscalls_assigned(0) { addSyscall(5, (void *)&User::irqListen); addSyscall(6, (void *)&User::panic); addSyscall(7, (void *)&User::registerName); - addSyscall(8, (void *)&User::registerNode); + addSyscall(8, (void *)&User::registerStream); addSyscall(9, (void *)&User::exit); - addSyscall(10, (void *)&User::obtainNodeWriter); - addSyscall(11, (void *)&User::obtainNodeListener); - addSyscall(12, (void *)&User::readNode); - addSyscall(13, (void *)&User::readNodeUnblock); - addSyscall(14, (void *)&User::writeNode); + addSyscall(10, (void *)&User::obtainStreamWriter); + addSyscall(11, (void *)&User::obtainStreamListener); + addSyscall(12, (void *)&User::readStream); + addSyscall(13, (void *)&User::readStreamUnblock); + addSyscall(14, (void *)&User::writeStream); addSyscall(15, (void *)&User::defer); addSyscall(16, (void *)&User::malloc); addSyscall(17, (void *)&User::free); addSyscall(18, (void *)&User::memcpy); + addSyscall(19, (void *)&User::registerQueue); } u8 Syscall::versionMajor() const { return 0; } Syscall.cpp @@ -234,10 +234,10 @@ void Tasks::Task::setIOMap(u8 port, bool enabled) { iomap[port / 8] |= (1 << (port % 8)); } -Tasks::Task::Node::Node(): _wl_id(0) { +Tasks::Task::Stream::Stream(): _wl_id(0) { } -u32 Tasks::Task::Node::registerWriter(bool exclusive) { +u32 Tasks::Task::Stream::registerWriter(bool exclusive) { if (_exclusive) return -1; if (exclusive) { @@ -253,13 +253,13 @@ u32 Tasks::Task::Node::registerWriter(bool exclusive) { return id; } -u32 Tasks::Task::Node::registerListener() { +u32 Tasks::Task::Stream::registerListener() { u32 id = _nextId(); _listeners.push_back(Listener(id)); return id; } -Tasks::Task::Node::Listener &Tasks::Task::Node::getListener(u32 id) { +Tasks::Task::Stream::Listener &Tasks::Task::Stream::getListener(u32 id) { for (LinkedList<Listener>::iterator it = _listeners.begin(); it != _listeners.end(); ++it) { if (it->_id == id) return *it; @@ -267,13 +267,13 @@ Tasks::Task::Node::Listener &Tasks::Task::Node::getListener(u32 id) { AkariPanic("No matching listener - TODO: something more useful."); } -void Tasks::Task::Node::writeAllListeners(const char *buffer, u32 n) { +void Tasks::Task::Stream::writeAllListeners(const char *buffer, u32 n) { for (LinkedList<Listener>::iterator it = _listeners.begin(); it != _listeners.end(); ++it) { it->append(buffer, n); } } -bool Tasks::Task::Node::hasWriter(u32 id) const { +bool Tasks::Task::Stream::hasWriter(u32 id) const { for (LinkedList<u32>::iterator it = _writers.begin(); it != _writers.end(); ++it) { if (*it == id) return true; @@ -281,7 +281,7 @@ bool Tasks::Task::Node::hasWriter(u32 id) const { return false; } -bool Tasks::Task::Node::hasListener(u32 id) const { +bool Tasks::Task::Stream::hasListener(u32 id) const { for (LinkedList<Listener>::iterator it = _listeners.begin(); it != _listeners.end(); ++it) { if (it->_id == id) return true; @@ -289,10 +289,10 @@ bool Tasks::Task::Node::hasListener(u32 id) const { return false; } -Tasks::Task::Node::Listener::Listener(u32 id): _id(id), _buffer(0), _buflen(0), _hooked(0) +Tasks::Task::Stream::Listener::Listener(u32 id): _id(id), _buffer(0), _buflen(0), _hooked(0) { } -void Tasks::Task::Node::Listener::append(const char *data, u32 n) { +void Tasks::Task::Stream::Listener::append(const char *data, u32 n) { // HACK: hacky little appending string reallocing stupid buffer. // Write a proper appending buffer (with smarts) and refactor it later. if (n == 0) return; @@ -315,7 +315,7 @@ void Tasks::Task::Node::Listener::append(const char *data, u32 n) { } } -void Tasks::Task::Node::Listener::reset() { +void Tasks::Task::Stream::Listener::reset() { if (_buffer) { delete [] _buffer; _buffer = 0; @@ -323,7 +323,7 @@ void Tasks::Task::Node::Listener::reset() { } } -void Tasks::Task::Node::Listener::cut(u32 n) { +void Tasks::Task::Stream::Listener::cut(u32 n) { if (!_buffer) return; if (_buflen <= n) { delete [] _buffer; @@ -341,26 +341,28 @@ void Tasks::Task::Node::Listener::cut(u32 n) { } } -void Tasks::Task::Node::Listener::hook(Task *task) { +void Tasks::Task::Stream::Listener::hook(Task *task) { _hooked = task; } -void Tasks::Task::Node::Listener::unhook() { +void Tasks::Task::Stream::Listener::unhook() { _hooked = 0; } -const char *Tasks::Task::Node::Listener::view() const { +const char *Tasks::Task::Stream::Listener::view() const { return _buffer; } -u32 Tasks::Task::Node::Listener::length() const { +u32 Tasks::Task::Stream::Listener::length() const { return _buflen; } -u32 Tasks::Task::Node::_nextId() { +u32 Tasks::Task::Stream::_nextId() { return ++_wl_id; } +Tasks::Task::Queue::Queue() { } + Tasks::Task::Task(u8 cpl): next(0), priorityNext(0), irqWaiting(false), irqListen(0), irqListenHits(0), userWaiting(false), userCall(0), @@ -374,5 +376,6 @@ Tasks::Task::Task(u8 cpl): for (u8 i = 0; i < 32; ++i) iomap[i] = 0xFF; - nodesByName = new HashTable<Symbol, Node *>(); + streamsByName = new HashTable<Symbol, Stream *>(); + queuesByName = new HashTable<Symbol, Queue *>(); } Tasks.cpp @@ -113,11 +113,10 @@ void ATAProcess() { */ // Now we need to wait and listen for commands! - // I think we need a message passing method for this, not just our current reader/writer. if (!SYSCALL_BOOL(syscall_registerName("system.io.ata"))) syscall_panic("could not register system.io.ata"); - if (!SYSCALL_BOOL(syscall_registerNode("command"))) + if (!SYSCALL_BOOL(syscall_registerQueue("command"))) syscall_panic("could not register system.io.ata:command"); } TmpATA.cpp @@ -7,6 +7,7 @@ #include <Tasks.hpp> #include <Syscall.hpp> + namespace User { void putc(char c) { Akari->console->putChar(c); @@ -58,8 +59,8 @@ namespace User { return true; } - bool registerNode(const char *node) { - Symbol sNode(node); + bool registerStream(const char *name) { + Symbol sNode(name); if (!Akari->tasks->current->registeredName) { // TODO: just kill the process, don't kill the system. // TODO: is this correct behaviour? Or could we have registered nodes @@ -67,13 +68,30 @@ namespace User { AkariPanic("name not registered - cannot register node"); } - if (Akari->tasks->current->nodesByName->hasKey(node)) { + if (Akari->tasks->current->streamsByName->hasKey(name)) { + AkariPanic("node already registered - cannot register atop it"); + } + + Tasks::Task::Stream *target = new Tasks::Task::Stream(); + + (*Akari->tasks->current->streamsByName)[sNode] = target; + return true; + } + + bool registerQueue(const char *name) { + Symbol sNode(name); + if (!Akari->tasks->current->registeredName) { + // TODO + AkariPanic("name not registered - cannot register node"); + } + + if (Akari->tasks->current->streamsByName->hasKey(name)) { AkariPanic("node already registered - cannot register atop it"); } - Tasks::Task::Node *target = new Tasks::Task::Node(); + Tasks::Task::Queue *target = new Tasks::Task::Queue(); - (*Akari->tasks->current->nodesByName)[sNode] = target; + (*Akari->tasks->current->queuesByName)[sNode] = target; return true; } @@ -95,33 +113,33 @@ namespace User { // Gone! XXX what happens when the last task exists!? Everything probably goes to hell ... } - static inline Tasks::Task::Node *getNode(const char *name, const char *node) { + static inline Tasks::Task::Stream *getStream(const char *name, const char *node) { Symbol sName(name), sNode(node); if (!Akari->tasks->registeredTasks->hasKey(sName)) return 0; Tasks::Task *task = (*Akari->tasks->registeredTasks)[sName]; - if (!task->nodesByName->hasKey(sNode)) + if (!task->streamsByName->hasKey(sNode)) return 0; - return (*task->nodesByName)[sNode]; + return (*task->streamsByName)[sNode]; } - u32 obtainNodeWriter(const char *name, const char *node, bool exclusive) { - Tasks::Task::Node *target = getNode(name, node); + u32 obtainStreamWriter(const char *name, const char *node, bool exclusive) { + Tasks::Task::Stream *target = getStream(name, node); if (!target) return -1; return target->registerWriter(exclusive); } - u32 obtainNodeListener(const char *name, const char *node) { - Tasks::Task::Node *target = getNode(name, node); + u32 obtainStreamListener(const char *name, const char *node) { + Tasks::Task::Stream *target = getStream(name, node); if (!target) return -1; return target->registerListener(); } // Keeping in mind that `buffer''s data probably isn't asciz. - u32 readNode_impl(const char *name, const char *node, u32 listener, char *buffer, u32 n, bool block) { + u32 readStream_impl(const char *name, const char *node, u32 listener, char *buffer, u32 n, bool block) { ReadCall c(name, node, listener, buffer, n); u32 r = c(); if (!block || !c.shallBlock()) @@ -129,7 +147,7 @@ namespace User { // block && r.shallBlock() // Block until such time as some data is available. - Tasks::Task::Node::Listener *l = c.getListener(); + Tasks::Task::Stream::Listener *l = c.getListener(); Akari->tasks->current->userWaiting = true; Akari->tasks->current->userCall = new ReadCall(c); @@ -138,15 +156,15 @@ namespace User { return 0; } - u32 readNode(const char *name, const char *node, u32 listener, char *buffer, u32 n) { - return readNode_impl(name, node, listener, buffer, n, true); + u32 readStream(const char *name, const char *node, u32 listener, char *buffer, u32 n) { + return readStream_impl(name, node, listener, buffer, n, true); } - u32 readNodeUnblock(const char *name, const char *node, u32 listener, char *buffer, u32 n) { - return readNode_impl(name, node, listener, buffer, n, false); + u32 readStreamUnblock(const char *name, const char *node, u32 listener, char *buffer, u32 n) { + return readStream_impl(name, node, listener, buffer, n, false); } - u32 writeNode(const char *name, const char *node, u32 writer, const char *buffer, u32 n) { - Tasks::Task::Node *target = getNode(name, node); + u32 writeStream(const char *name, const char *node, u32 writer, const char *buffer, u32 n) { + Tasks::Task::Stream *target = getStream(name, node); if (!target || !target->hasWriter(writer)) return -1; // We do have a writer, so we can go ahead and write to all listeners. @@ -177,11 +195,11 @@ namespace User { return dest; } - ReadCall::ReadCall(const char *name, const char *node, u32 listener, char *buffer, u32 n): - _listener(&getNode(name, node)->getListener(listener)), _buffer(buffer), _n(n) + ReadCall::ReadCall(const char *name, const char *node, u32 listener, char *buffer, u32 n): + _listener(&getStream(name, node)->getListener(listener)), _buffer(buffer), _n(n) { } - Tasks::Task::Node::Listener *ReadCall::getListener() const { + Tasks::Task::Stream::Listener *ReadCall::getListener() const { return _listener; } UserCalls.cpp @@ -63,3 +63,4 @@ DEFN_SYSCALL0(defer, 15); DEFN_SYSCALL1(malloc, 16, u32); DEFN_SYSCALL1(free, 17, void *); DEFN_SYSCALL3(memcpy, 18, void *, const void *, u32); +DEFN_SYSCALL1(registerQueue, 19, const char *); UserGates.cpp @@ -30,9 +30,10 @@ public: class Task { public: - class Node { + + class Stream { public: - Node(); + Stream(); class Listener; u32 registerWriter(bool exclusive); @@ -44,7 +45,7 @@ public: bool hasListener(u32 id) const; class Listener { - friend class Node; + friend class Stream; public: Listener(u32 id); @@ -73,6 +74,12 @@ public: u32 _nextId(); }; + class Queue { + public: + Queue(); + + }; + static Task *BootstrapInitialTask(u8 cpl, Memory::PageDirectory *pageDirBase); static Task *CreateTask(u32 entry, u8 cpl, bool interruptFlag, u8 iopl, Memory::PageDirectory *pageDirBase); @@ -102,7 +109,8 @@ public: u32 ks; u8 iomap[32]; - HashTable<Symbol, Node *> *nodesByName; + HashTable<Symbol, Stream *> *streamsByName; + HashTable<Symbol, Queue *> *queuesByName; protected: Task(u8 cpl); inc/Tasks.hpp @@ -15,13 +15,14 @@ namespace User { void irqListen(u32 irq); void panic(const char *s); bool registerName(const char *name); - bool registerNode(const char *name); + bool registerStream(const char *name); + bool registerQueue(const char *name); void exit(); - u32 obtainNodeWriter(const char *name, const char *node, bool exclusive); - u32 obtainNodeListener(const char *name, const char *node); - u32 readNode(const char *name, const char *node, u32 listener, char *buffer, u32 n); - u32 readNodeUnblock(const char *name, const char *node, u32 listener, char *buffer, u32 n); - u32 writeNode(const char *name, const char *node, u32 writer, const char *buffer, u32 n); + u32 obtainStreamWriter(const char *name, const char *node, bool exclusive); + u32 obtainStreamListener(const char *name, const char *node); + u32 readStream(const char *name, const char *node, u32 listener, char *buffer, u32 n); + u32 readStreamUnblock(const char *name, const char *node, u32 listener, char *buffer, u32 n); + u32 writeStream(const char *name, const char *node, u32 writer, const char *buffer, u32 n); void defer(); void *malloc(u32 n); void free(void *p); @@ -31,12 +32,12 @@ namespace User { public: ReadCall(const char *name, const char *node, u32 listener, char *buffer, u32 n); - Tasks::Task::Node::Listener *getListener() const; + Tasks::Task::Stream::Listener *getListener() const; u32 operator ()(); protected: - Tasks::Task::Node::Listener *_listener; + Tasks::Task::Stream::Listener *_listener; char *_buffer; u32 _n; }; inc/UserCalls.hpp @@ -31,6 +31,7 @@ DECL_SYSCALL0(defer); DECL_SYSCALL1(malloc, u32); DECL_SYSCALL1(free, void *); DECL_SYSCALL3(memcpy, void *, const void *, u32); +DECL_SYSCALL1(registerQueue, const char *); #define SYSCALL_BOOL(x) ((bool)((x) & 0xFF)) inc/UserGates.hpp 2e1976e2030f1f531bd057728b4e48bdead4c7fd Arlen Cuss celtic@sairyx.org http://github.com/celtic/akari/commit/eaed73dbbcc8cfb778edec21e8642a29cd4ce933 eaed73dbbcc8cfb778edec21e8642a29cd4ce933 2009-11-06T20:25:37-08:00 2009-11-06T20:25:37-08:00 After much deliberation, streams and queues ... f4cbc3897f05eb31b1eace1846ba0e14b4e55c74 Arlen Cuss celtic@sairyx.org