Index: src/Makefile ================================================================== --- src/Makefile +++ src/Makefile @@ -90,20 +90,19 @@ OFINIFile.m \ OFSettings.m \ OFZIPArchive.m \ OFZIPArchiveEntry.m SRCS_PLUGINS = OFPlugin.m -SRCS_SOCKETS = OFHTTPClient.m \ - OFHTTPRequest.m \ - OFHTTPResponse.m \ - OFHTTPServer.m \ - OFKernelEventObserver.m \ - OFKernelEventObserver_LockedQueue.m \ - OFStreamSocket.m \ - OFTCPSocket.m \ - OFUDPSocket.m \ - resolver.m \ +SRCS_SOCKETS = OFHTTPClient.m \ + OFHTTPRequest.m \ + OFHTTPResponse.m \ + OFHTTPServer.m \ + OFKernelEventObserver.m \ + OFStreamSocket.m \ + OFTCPSocket.m \ + OFUDPSocket.m \ + resolver.m \ socket.m SRCS_THREADS = OFCondition.m \ OFMutex.m \ OFRecursiveMutex.m \ OFThreadPool.m \ Index: src/OFKernelEventObserver+Private.h ================================================================== --- src/OFKernelEventObserver+Private.h +++ src/OFKernelEventObserver+Private.h @@ -17,9 +17,14 @@ #import "OFKernelEventObserver.h" OF_ASSUME_NONNULL_BEGIN @interface OFKernelEventObserver () +- (void)OF_addObjectForReading: (id )object; +- (void)OF_addObjectForWriting: (id )object; +- (void)OF_removeObjectForReading: (id )object; +- (void)OF_removeObjectForWriting: (id )object; +- (void)OF_processQueue; - (bool)OF_processReadBuffers; @end OF_ASSUME_NONNULL_END Index: src/OFKernelEventObserver.h ================================================================== --- src/OFKernelEventObserver.h +++ src/OFKernelEventObserver.h @@ -23,10 +23,11 @@ @class OFMutableArray OF_GENERIC(ObjectType); @class OFDate; #ifdef OF_HAVE_THREADS @class OFMutex; #endif +@class OFDataArray; /*! * @protocol OFKernelEventObserverDelegate * OFKernelEventObserver.h ObjFW/OFKernelEventObserver.h * @@ -118,10 +119,12 @@ struct sockaddr_in _cancelAddr; #endif #ifdef OF_HAVE_THREADS OFMutex *_mutex; #endif + OFDataArray *_queueActions; + OFMutableArray *_queueObjects; } /*! * The delegate for the OFKernelEventObserver. */ Index: src/OFKernelEventObserver.m ================================================================== --- src/OFKernelEventObserver.m +++ src/OFKernelEventObserver.m @@ -19,10 +19,11 @@ #include "config.h" #import "OFKernelEventObserver.h" #import "OFKernelEventObserver+Private.h" #import "OFArray.h" +#import "OFDataArray.h" #import "OFStream.h" #import "OFStream+Private.h" #ifndef OF_HAVE_PIPE # import "OFStreamSocket.h" #endif @@ -49,10 +50,18 @@ #import "OFOutOfRangeException.h" #import "socket.h" #import "socket_helpers.h" +enum { + QUEUE_ADD = 0, + QUEUE_REMOVE = 1, + QUEUE_READ = 0, + QUEUE_WRITE = 2 +}; +#define QUEUE_ACTION (QUEUE_ADD | QUEUE_REMOVE) + @implementation OFKernelEventObserver @synthesize delegate = _delegate; + (void)initialize { @@ -135,10 +144,14 @@ #endif #ifdef OF_HAVE_THREADS _mutex = [[OFMutex alloc] init]; #endif + + _queueActions = [[OFDataArray alloc] + initWithItemSize: sizeof(int)]; + _queueObjects = [[OFMutableArray alloc] init]; } @catch (id e) { [self release]; @throw e; } @@ -160,32 +173,211 @@ #ifdef OF_HAVE_THREADS [_mutex release]; #endif + [_queueActions release]; + [_queueObjects release]; + [super dealloc]; } - (void)addObjectForReading: (id )object { - OF_UNRECOGNIZED_SELECTOR +#ifdef OF_HAVE_THREADS + [_mutex lock]; + @try { +#endif + int action = QUEUE_ADD | QUEUE_READ; + + [_queueActions addItem: &action]; + [_queueObjects addObject: object]; +#ifdef OF_HAVE_THREADS + } @finally { + [_mutex unlock]; + } +#endif + + [self cancel]; } - (void)addObjectForWriting: (id )object { - OF_UNRECOGNIZED_SELECTOR +#ifdef OF_HAVE_THREADS + [_mutex lock]; + @try { +#endif + int action = QUEUE_ADD | QUEUE_WRITE; + + [_queueActions addItem: &action]; + [_queueObjects addObject: object]; +#ifdef OF_HAVE_THREADS + } @finally { + [_mutex unlock]; + } +#endif + + [self cancel]; } - (void)removeObjectForReading: (id )object { - OF_UNRECOGNIZED_SELECTOR +#ifdef OF_HAVE_THREADS + [_mutex lock]; + @try { +#endif + int action = QUEUE_REMOVE | QUEUE_READ; + + [_queueActions addItem: &action]; + [_queueObjects addObject: object]; +#ifdef OF_HAVE_THREADS + } @finally { + [_mutex unlock]; + } +#endif + + [self cancel]; } - (void)removeObjectForWriting: (id )object { +#ifdef OF_HAVE_THREADS + [_mutex lock]; + @try { +#endif + int action = QUEUE_REMOVE | QUEUE_WRITE; + + [_queueActions addItem: &action]; + [_queueObjects addObject: object]; +#ifdef OF_HAVE_THREADS + } @finally { + [_mutex unlock]; + } +#endif + + [self cancel]; +} + +- (void)OF_addObjectForReading: (id )object +{ + OF_UNRECOGNIZED_SELECTOR +} + +- (void)OF_addObjectForWriting: (id )object +{ + OF_UNRECOGNIZED_SELECTOR +} + +- (void)OF_removeObjectForReading: (id )object +{ + OF_UNRECOGNIZED_SELECTOR +} + +- (void)OF_removeObjectForWriting: (id )object +{ OF_UNRECOGNIZED_SELECTOR } + +- (void)OF_processQueue +{ + void *pool = objc_autoreleasePoolPush(); + +#ifdef OF_HAVE_THREADS + [_mutex lock]; + @try { +#endif + int *queueActions = [_queueActions items]; + id const *queueObjects = [_queueObjects objects]; + size_t count = [_queueActions count]; + + OF_ENSURE([_queueObjects count] == count); + + for (size_t i = 0; i < count; i++) { + int action = queueActions[i]; + id object = queueObjects[i]; + + switch (action) { + case QUEUE_ADD | QUEUE_READ: + [_readObjects addObject: object]; + + @try { + [self OF_addObjectForReading: object]; + } @catch (id e) { + [_readObjects + removeObjectIdenticalTo: object]; + + @throw e; + } + + break; + case QUEUE_ADD | QUEUE_WRITE: + [_writeObjects addObject: object]; + + @try { + [self OF_addObjectForWriting: object]; + } @catch (id e) { + [_writeObjects + removeObjectIdenticalTo: object]; + + @throw e; + } + + break; + case QUEUE_REMOVE | QUEUE_READ: + [self OF_removeObjectForReading: object]; + + [_readObjects removeObjectIdenticalTo: object]; + + break; + case QUEUE_REMOVE | QUEUE_WRITE: + [self OF_removeObjectForWriting: object]; + + [_writeObjects removeObjectIdenticalTo: object]; + + break; + default: + OF_ENSURE(0); + } + } + + [_queueActions removeAllItems]; + [_queueObjects removeAllObjects]; +#ifdef OF_HAVE_THREADS + } @finally { + [_mutex unlock]; + } +#endif + + objc_autoreleasePoolPop(pool); +} + +- (bool)OF_processReadBuffers +{ + bool foundInReadBuffer = false; + + for (id object in _readObjects) { + void *pool = objc_autoreleasePoolPush(); + + if ([object isKindOfClass: [OFStream class]] && + [object hasDataInReadBuffer] && + ![object OF_isWaitingForDelimiter]) { + if ([_delegate respondsToSelector: + @selector(objectIsReadyForReading:)]) + [_delegate objectIsReadyForReading: object]; + + foundInReadBuffer = true; + } + + objc_autoreleasePoolPop(pool); + } + + /* + * As long as we have data in the read buffer for any stream, we don't + * want to block. + */ + return foundInReadBuffer; +} - (void)observe { [self observeForTimeInterval: -1]; } @@ -212,33 +404,6 @@ OF_ENSURE(sendto(_cancelFD[1], "", 1, 0, (struct sockaddr*)&_cancelAddr, 8) > 0); # endif #endif } - -- (bool)OF_processReadBuffers -{ - bool foundInReadBuffer = false; - - for (id object in _readObjects) { - void *pool = objc_autoreleasePoolPush(); - - if ([object isKindOfClass: [OFStream class]] && - [object hasDataInReadBuffer] && - ![object OF_isWaitingForDelimiter]) { - if ([_delegate respondsToSelector: - @selector(objectIsReadyForReading:)]) - [_delegate objectIsReadyForReading: object]; - - foundInReadBuffer = true; - } - - objc_autoreleasePoolPop(pool); - } - - /* - * As long as we have data in the read buffer for any stream, we don't - * want to block. - */ - return foundInReadBuffer; -} @end DELETED src/OFKernelEventObserver_LockedQueue.h Index: src/OFKernelEventObserver_LockedQueue.h ================================================================== --- src/OFKernelEventObserver_LockedQueue.h +++ src/OFKernelEventObserver_LockedQueue.h @@ -1,41 +0,0 @@ -/* - * Copyright (c) 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016 - * Jonathan Schleifer - * - * All rights reserved. - * - * This file is part of ObjFW. It may be distributed under the terms of the - * Q Public License 1.0, which can be found in the file LICENSE.QPL included in - * the packaging of this file. - * - * Alternatively, it may be distributed under the terms of the GNU General - * Public License, either version 2 or 3, which can be found in the file - * LICENSE.GPLv2 or LICENSE.GPLv3 respectively included in the packaging of this - * file. - */ - -#import "OFKernelEventObserver.h" - -OF_ASSUME_NONNULL_BEGIN - -@class OFMutableArray OF_GENERIC(ObjectType); -@class OFDataArray; - -@interface OFKernelEventObserver_LockedQueue: OFKernelEventObserver -{ - OFDataArray *_queueActions, *_queueFDs; - OFMutableArray *_queueObjects; -} - -- (void)OF_addObjectForReading: (id)object - fileDescriptor: (int)fd; -- (void)OF_addObjectForWriting: (id)object - fileDescriptor: (int)fd; -- (void)OF_removeObjectForReading: (id)object - fileDescriptor: (int)fd; -- (void)OF_removeObjectForWriting: (id)object - fileDescriptor: (int)fd; -- (void)OF_processQueue; -@end - -OF_ASSUME_NONNULL_END DELETED src/OFKernelEventObserver_LockedQueue.m Index: src/OFKernelEventObserver_LockedQueue.m ================================================================== --- src/OFKernelEventObserver_LockedQueue.m +++ src/OFKernelEventObserver_LockedQueue.m @@ -1,237 +0,0 @@ -/* - * Copyright (c) 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016 - * Jonathan Schleifer - * - * All rights reserved. - * - * This file is part of ObjFW. It may be distributed under the terms of the - * Q Public License 1.0, which can be found in the file LICENSE.QPL included in - * the packaging of this file. - * - * Alternatively, it may be distributed under the terms of the GNU General - * Public License, either version 2 or 3, which can be found in the file - * LICENSE.GPLv2 or LICENSE.GPLv3 respectively included in the packaging of this - * file. - */ - -#include "config.h" - -#import "OFKernelEventObserver_LockedQueue.h" -#import "OFArray.h" -#import "OFDataArray.h" -#ifdef OF_HAVE_THREADS -# import "OFMutex.h" -#endif - -#import "OFInitializationFailedException.h" - -enum { - QUEUE_ADD = 0, - QUEUE_REMOVE = 1, - QUEUE_READ = 0, - QUEUE_WRITE = 2 -}; -#define QUEUE_ACTION (QUEUE_ADD | QUEUE_REMOVE) - -@implementation OFKernelEventObserver_LockedQueue -- init -{ - self = [super init]; - - @try { - _queueActions = [[OFDataArray alloc] - initWithItemSize: sizeof(int)]; - _queueFDs = [[OFDataArray alloc] initWithItemSize: sizeof(int)]; - _queueObjects = [[OFMutableArray alloc] init]; - } @catch (id e) { - @throw [OFInitializationFailedException - exceptionWithClass: [self class]]; - } - - return self; -} - -- (void)dealloc -{ - [_queueActions release]; - [_queueFDs release]; - [_queueObjects release]; - - [super dealloc]; -} - -- (void)addObjectForReading: (id )object -{ -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - int qi = QUEUE_ADD | QUEUE_READ; - int fd = [object fileDescriptorForReading]; - - [_queueActions addItem: &qi]; - [_queueFDs addItem: &fd]; - [_queueObjects addObject: object]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif - - [self cancel]; -} - -- (void)addObjectForWriting: (id )object -{ -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - int qi = QUEUE_ADD | QUEUE_WRITE; - int fd = [object fileDescriptorForWriting]; - - [_queueActions addItem: &qi]; - [_queueFDs addItem: &fd]; - [_queueObjects addObject: object]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif - - [self cancel]; -} - -- (void)removeObjectForReading: (id )object -{ -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - int qi = QUEUE_REMOVE | QUEUE_READ; - int fd = [object fileDescriptorForReading]; - - [_queueActions addItem: &qi]; - [_queueFDs addItem: &fd]; - [_queueObjects addObject: object]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif - - [self cancel]; -} - -- (void)removeObjectForWriting: (id )object -{ -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - int qi = QUEUE_REMOVE | QUEUE_WRITE; - int fd = [object fileDescriptorForWriting]; - - [_queueActions addItem: &qi]; - [_queueFDs addItem: &fd]; - [_queueObjects addObject: object]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif - - [self cancel]; -} - -- (void)OF_addObjectForReading: (id)object - fileDescriptor: (int)fd -{ - OF_UNRECOGNIZED_SELECTOR -} - -- (void)OF_addObjectForWriting: (id)object - fileDescriptor: (int)fd -{ - OF_UNRECOGNIZED_SELECTOR -} - -- (void)OF_removeObjectForReading: (id)object - fileDescriptor: (int)fd -{ - OF_UNRECOGNIZED_SELECTOR -} - -- (void)OF_removeObjectForWriting: (id)object - fileDescriptor: (int)fd -{ - OF_UNRECOGNIZED_SELECTOR -} - -- (void)OF_processQueue -{ - void *pool = objc_autoreleasePoolPush(); - -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - int *queueActions = [_queueActions items]; - int *queueFDs = [_queueFDs items]; - id const *queueObjects = [_queueObjects objects]; - size_t count = [_queueActions count]; - - OF_ENSURE([_queueFDs count] == count); - OF_ENSURE([_queueObjects count] == count); - - for (size_t i = 0; i < count; i++) { - int action = queueActions[i]; - int fd = queueFDs[i]; - id object = queueObjects[i]; - - switch (action) { - case QUEUE_ADD | QUEUE_READ: - [_readObjects addObject: object]; - - [self OF_addObjectForReading: object - fileDescriptor: fd]; - - break; - case QUEUE_ADD | QUEUE_WRITE: - [_writeObjects addObject: object]; - - [self OF_addObjectForWriting: object - fileDescriptor: fd]; - - break; - case QUEUE_REMOVE | QUEUE_READ: - [self OF_removeObjectForReading: object - fileDescriptor: fd]; - - [_readObjects removeObjectIdenticalTo: object]; - - break; - case QUEUE_REMOVE | QUEUE_WRITE: - [self OF_removeObjectForWriting: object - fileDescriptor: fd]; - - [_writeObjects removeObjectIdenticalTo: object]; - - break; - default: - OF_ENSURE(0); - } - } - - [_queueActions removeAllItems]; - [_queueFDs removeAllItems]; - [_queueObjects removeAllObjects]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif - - objc_autoreleasePoolPop(pool); -} -@end Index: src/OFKernelEventObserver_epoll.m ================================================================== --- src/OFKernelEventObserver_epoll.m +++ src/OFKernelEventObserver_epoll.m @@ -90,129 +90,97 @@ } - (void)OF_addObject: (id)object fileDescriptor: (int)fd events: (int)addEvents - objectsArray: (OFMutableArray*)objectsArray -{ -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - struct epoll_event event; - intptr_t events; - - events = (intptr_t)[_FDToEvents - valueForKey: (void*)(intptr_t)fd]; - - memset(&event, 0, sizeof(event)); - event.events = (int)events | addEvents; - event.data.ptr = object; - - [objectsArray addObject: object]; - - if (epoll_ctl(_epfd, - (events == 0 ? EPOLL_CTL_ADD : EPOLL_CTL_MOD), - fd, &event) == -1) { - [objectsArray removeObjectIdenticalTo: object]; - @throw [OFObserveFailedException - exceptionWithObserver: self - errNo: errno]; - } - - [_FDToEvents setValue: (void*)(events | addEvents) - forKey: (void*)(intptr_t)fd]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif +{ + struct epoll_event event; + intptr_t events; + + events = (intptr_t)[_FDToEvents valueForKey: (void*)(intptr_t)fd]; + + memset(&event, 0, sizeof(event)); + event.events = (int)events | addEvents; + event.data.ptr = object; + + if (epoll_ctl(_epfd, (events == 0 ? EPOLL_CTL_ADD : EPOLL_CTL_MOD), + fd, &event) == -1) + @throw [OFObserveFailedException exceptionWithObserver: self + errNo: errno]; + + [_FDToEvents setValue: (void*)(events | addEvents) + forKey: (void*)(intptr_t)fd]; } - (void)OF_removeObject: (id)object fileDescriptor: (int)fd events: (int)removeEvents - objectsArray: (OFMutableArray*)objectsArray -{ -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - intptr_t events; - - events = (intptr_t)[_FDToEvents - valueForKey: (void*)(intptr_t)fd]; - events &= ~removeEvents; - - if (events == 0) { - if (epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, NULL) == -1) - @throw [OFObserveFailedException - exceptionWithObserver: self - errNo: errno]; - - [_FDToEvents removeValueForKey: (void*)(intptr_t)fd]; - } else { - struct epoll_event event; - - memset(&event, 0, sizeof(event)); - event.events = (int)events; - event.data.ptr = object; - - if (epoll_ctl(_epfd, EPOLL_CTL_MOD, fd, &event) == -1) - @throw [OFObserveFailedException - exceptionWithObserver: self - errNo: errno]; - - [_FDToEvents setValue: (void*)events - forKey: (void*)(intptr_t)fd]; - } - - [objectsArray removeObjectIdenticalTo: object]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif -} - -- (void)addObjectForReading: (id )object -{ - [self OF_addObject: object - fileDescriptor: [object fileDescriptorForReading] - events: EPOLLIN - objectsArray: _readObjects]; -} - -- (void)addObjectForWriting: (id )object -{ - [self OF_addObject: object - fileDescriptor: [object fileDescriptorForWriting] - events: EPOLLOUT - objectsArray: _writeObjects]; -} - -- (void)removeObjectForReading: (id )object -{ - [self OF_removeObject: object - fileDescriptor: [object fileDescriptorForReading] - events: EPOLLIN - objectsArray: _readObjects]; -} - -- (void)removeObjectForWriting: (id )object -{ - [self OF_removeObject: object - fileDescriptor: [object fileDescriptorForWriting] - events: EPOLLOUT - objectsArray: _writeObjects]; +{ + intptr_t events; + + events = (intptr_t)[_FDToEvents valueForKey: (void*)(intptr_t)fd]; + events &= ~removeEvents; + + if (events == 0) { + if (epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, NULL) == -1) + @throw [OFObserveFailedException + exceptionWithObserver: self + errNo: errno]; + + [_FDToEvents removeValueForKey: (void*)(intptr_t)fd]; + } else { + struct epoll_event event; + + memset(&event, 0, sizeof(event)); + event.events = (int)events; + event.data.ptr = object; + + if (epoll_ctl(_epfd, EPOLL_CTL_MOD, fd, &event) == -1) + @throw [OFObserveFailedException + exceptionWithObserver: self + errNo: errno]; + + [_FDToEvents setValue: (void*)events + forKey: (void*)(intptr_t)fd]; + } +} + +- (void)OF_addObjectForReading: (id )object +{ + [self OF_addObject: object + fileDescriptor: [object fileDescriptorForReading] + events: EPOLLIN]; +} + +- (void)OF_addObjectForWriting: (id )object +{ + [self OF_addObject: object + fileDescriptor: [object fileDescriptorForWriting] + events: EPOLLOUT]; +} + +- (void)OF_removeObjectForReading: (id )object +{ + [self OF_removeObject: object + fileDescriptor: [object fileDescriptorForReading] + events: EPOLLIN]; +} + +- (void)OF_removeObjectForWriting: (id )object +{ + [self OF_removeObject: object + fileDescriptor: [object fileDescriptorForWriting] + events: EPOLLOUT]; } - (void)observeForTimeInterval: (of_time_interval_t)timeInterval { OFNull *nullObject = [OFNull null]; struct epoll_event eventList[EVENTLIST_SIZE]; int events; + + [self OF_processQueue]; if ([self OF_processReadBuffers]) return; events = epoll_wait(_epfd, eventList, EVENTLIST_SIZE, @@ -231,11 +199,11 @@ continue; } if (eventList[i].events & EPOLLIN) { - pool = objc_autoreleasePoolPush(); + void *pool = objc_autoreleasePoolPush(); if ([_delegate respondsToSelector: @selector(objectIsReadyForReading:)]) [_delegate objectIsReadyForReading: eventList[i].data.ptr]; @@ -242,11 +210,11 @@ objc_autoreleasePoolPop(pool); } if (eventList[i].events & EPOLLOUT) { - pool = objc_autoreleasePoolPush(); + void *pool = objc_autoreleasePoolPush(); if ([_delegate respondsToSelector: @selector(objectIsReadyForWriting:)]) [_delegate objectIsReadyForWriting: eventList[i].data.ptr]; Index: src/OFKernelEventObserver_kqueue.m ================================================================== --- src/OFKernelEventObserver_kqueue.m +++ src/OFKernelEventObserver_kqueue.m @@ -83,11 +83,11 @@ close(_kernelQueue); [super dealloc]; } -- (void)addObjectForReading: (id )object +- (void)OF_addObjectForReading: (id )object { struct kevent event; memset(&event, 0, sizeof(event)); event.ident = [object fileDescriptorForReading]; @@ -97,30 +97,16 @@ event.udata = object; #else event.udata = (intptr_t)object; #endif -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - [_readObjects addObject: object]; - - if (kevent(_kernelQueue, &event, 1, NULL, 0, NULL) != 0) { - [_readObjects removeObjectIdenticalTo: object]; - @throw [OFObserveFailedException - exceptionWithObserver: self - errNo: errno]; - } -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif -} - -- (void)addObjectForWriting: (id )object + if (kevent(_kernelQueue, &event, 1, NULL, 0, NULL) != 0) + @throw [OFObserveFailedException exceptionWithObserver: self + errNo: errno]; +} + +- (void)OF_addObjectForWriting: (id )object { struct kevent event; memset(&event, 0, sizeof(event)); event.ident = [object fileDescriptorForWriting]; @@ -130,86 +116,50 @@ event.udata = object; #else event.udata = (intptr_t)object; #endif -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - [_writeObjects addObject: object]; - - if (kevent(_kernelQueue, &event, 1, NULL, 0, NULL) != 0) { - [_writeObjects removeObjectIdenticalTo: object]; - @throw [OFObserveFailedException - exceptionWithObserver: self - errNo: errno]; - } -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif -} - -- (void)removeObjectForReading: (id )object + if (kevent(_kernelQueue, &event, 1, NULL, 0, NULL) != 0) + @throw [OFObserveFailedException exceptionWithObserver: self + errNo: errno]; +} + +- (void)OF_removeObjectForReading: (id )object { struct kevent event; memset(&event, 0, sizeof(event)); event.ident = [object fileDescriptorForReading]; event.filter = EVFILT_READ; event.flags = EV_DELETE; -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - if (kevent(_kernelQueue, &event, 1, NULL, 0, NULL) != 0) - @throw [OFObserveFailedException - exceptionWithObserver: self - errNo: errno]; - - [_readObjects removeObjectIdenticalTo: object]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif -} - -- (void)removeObjectForWriting: (id )object + if (kevent(_kernelQueue, &event, 1, NULL, 0, NULL) != 0) + @throw [OFObserveFailedException exceptionWithObserver: self + errNo: errno]; +} + +- (void)OF_removeObjectForWriting: (id )object { struct kevent event; memset(&event, 0, sizeof(event)); event.ident = [object fileDescriptorForWriting]; event.filter = EVFILT_WRITE; event.flags = EV_DELETE; -#ifdef OF_HAVE_THREADS - [_mutex lock]; - @try { -#endif - if (kevent(_kernelQueue, &event, 1, NULL, 0, NULL) != 0) - @throw [OFObserveFailedException - exceptionWithObserver: self - errNo: errno]; - - [_writeObjects removeObjectIdenticalTo: object]; -#ifdef OF_HAVE_THREADS - } @finally { - [_mutex unlock]; - } -#endif + if (kevent(_kernelQueue, &event, 1, NULL, 0, NULL) != 0) + @throw [OFObserveFailedException exceptionWithObserver: self + errNo: errno]; } - (void)observeForTimeInterval: (of_time_interval_t)timeInterval { struct timespec timeout; struct kevent eventList[EVENTLIST_SIZE]; int events; + + [self OF_processQueue]; if ([self OF_processReadBuffers]) return; timeout.tv_sec = (time_t)timeInterval; Index: src/OFKernelEventObserver_poll.h ================================================================== --- src/OFKernelEventObserver_poll.h +++ src/OFKernelEventObserver_poll.h @@ -12,20 +12,20 @@ * Public License, either version 2 or 3, which can be found in the file * LICENSE.GPLv2 or LICENSE.GPLv3 respectively included in the packaging of this * file. */ -#import "OFKernelEventObserver_LockedQueue.h" +#import "OFKernelEventObserver.h" OF_ASSUME_NONNULL_BEGIN @class OFDataArray; -@interface OFKernelEventObserver_poll: OFKernelEventObserver_LockedQueue +@interface OFKernelEventObserver_poll: OFKernelEventObserver { OFDataArray *_FDs; size_t _maxFD; id __unsafe_unretained *_FDToObject; } @end OF_ASSUME_NONNULL_END Index: src/OFKernelEventObserver_poll.m ================================================================== --- src/OFKernelEventObserver_poll.m +++ src/OFKernelEventObserver_poll.m @@ -123,39 +123,35 @@ break; } } } -- (void)OF_addObjectForReading: (id)object - fileDescriptor: (int)fd +- (void)OF_addObjectForReading: (id )object { [self OF_addObject: object - fileDescriptor: fd + fileDescriptor: [object fileDescriptorForReading] events: POLLIN]; } -- (void)OF_addObjectForWriting: (id)object - fileDescriptor: (int)fd +- (void)OF_addObjectForWriting: (id )object { [self OF_addObject: object - fileDescriptor: fd + fileDescriptor: [object fileDescriptorForWriting] events: POLLOUT]; } -- (void)OF_removeObjectForReading: (id)object - fileDescriptor: (int)fd +- (void)OF_removeObjectForReading: (id )object { [self OF_removeObject: object - fileDescriptor: fd + fileDescriptor: [object fileDescriptorForReading] events: POLLIN]; } -- (void)OF_removeObjectForWriting: (id)object - fileDescriptor: (int)fd +- (void)OF_removeObjectForWriting: (id )object { [self OF_removeObject: object - fileDescriptor: fd + fileDescriptor: [object fileDescriptorForWriting] events: POLLOUT]; } - (void)observeForTimeInterval: (of_time_interval_t)timeInterval { Index: src/OFKernelEventObserver_select.h ================================================================== --- src/OFKernelEventObserver_select.h +++ src/OFKernelEventObserver_select.h @@ -25,17 +25,17 @@ #ifdef HAVE_SYS_SELECT_H # include #endif -#import "OFKernelEventObserver_LockedQueue.h" +#import "OFKernelEventObserver.h" OF_ASSUME_NONNULL_BEGIN -@interface OFKernelEventObserver_select: OFKernelEventObserver_LockedQueue +@interface OFKernelEventObserver_select: OFKernelEventObserver { fd_set _readFDs, _writeFDs; int _maxFD; } @end OF_ASSUME_NONNULL_END Index: src/OFKernelEventObserver_select.m ================================================================== --- src/OFKernelEventObserver_select.m +++ src/OFKernelEventObserver_select.m @@ -62,13 +62,14 @@ _maxFD = (int)_cancelFD[0]; return self; } -- (void)OF_addObjectForReading: (id)object - fileDescriptor: (int)fd +- (void)OF_addObjectForReading: (id )object { + int fd = [object fileDescriptorForReading]; + if (fd < 0 || fd > INT_MAX - 1) @throw [OFOutOfRangeException exception]; #ifndef OF_WINDOWS if (fd >= FD_SETSIZE) @@ -79,13 +80,14 @@ _maxFD = fd; FD_SET(fd, &_readFDs); } -- (void)OF_addObjectForWriting: (id)object - fileDescriptor: (int)fd +- (void)OF_addObjectForWriting: (id )object { + int fd = [object fileDescriptorForWriting]; + if (fd < 0 || fd > INT_MAX - 1) @throw [OFOutOfRangeException exception]; #ifndef OF_WINDOWS if (fd >= FD_SETSIZE) @@ -96,14 +98,15 @@ _maxFD = fd; FD_SET(fd, &_writeFDs); } -- (void)OF_removeObjectForReading: (id)object - fileDescriptor: (int)fd +- (void)OF_removeObjectForReading: (id )object { /* TODO: Adjust _maxFD */ + + int fd = [object fileDescriptorForReading]; if (fd < 0) @throw [OFOutOfRangeException exception]; #ifndef OF_WINDOWS @@ -112,14 +115,15 @@ #endif FD_CLR(fd, &_readFDs); } -- (void)OF_removeObjectForWriting: (id)object - fileDescriptor: (int)fd +- (void)OF_removeObjectForWriting: (id )object { /* TODO: Adjust _maxFD */ + + int fd = [object fileDescriptorForWriting]; if (fd < 0) @throw [OFOutOfRangeException exception]; #ifndef OF_WINDOWS