diff --git a/AGENTS.md b/AGENTS.md index 4410dc5..ee12d1a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -52,6 +52,9 @@ beside this repository, and they consume the **installed** copy from `/usr/local initializer), then fills the instance inside `if (self) { ... }`, and returns `self`. - `dealloc` releases what the instance owns, then ends with `super(Object, self, dealloc)`. - `release` always returns `NULL`, so clear a field with `self->foo = release(self->foo)`. +- The members of an instance struct, after the parent struct and `interface[0]`, are in + alphabetical order. A private member is an ordinary member whose documentation carries + `@private`. A type MUST NOT nest its private members in a `locals` struct. - A `.c` file groups its methods under `#pragma mark - `, `#pragma mark - ` and `#pragma mark - Class lifecycle`. Methods in an interface struct and in its `.c` file are in alphabetical order. A private helper sits directly above its first caller. diff --git a/Documentation/index.md b/Documentation/index.md index 580444b..913e6fa 100644 --- a/Documentation/index.md +++ b/Documentation/index.md @@ -18,7 +18,7 @@ Object oriented framework for C. - **Unicode strings** with multibyte character set support - **Collections** for Objects and C types: Array, Dictionary, List, Set and more - **JSON** parsing, marshaling, and JSONPath queries -- **Concurrency**: Lock, Condition, Thread, Operation, OperationQueue +- **Concurrency**: Lock, Condition, Thread, ThreadPool, Operation, OperationQueue - **Networking**: URLSession, JSONContext & RESTClient ## tl;dr diff --git a/Objectively.vs15/Objectively.vcxproj b/Objectively.vs15/Objectively.vcxproj index 952f834..672172b 100644 --- a/Objectively.vs15/Objectively.vcxproj +++ b/Objectively.vs15/Objectively.vcxproj @@ -56,6 +56,7 @@ + @@ -114,6 +115,7 @@ + diff --git a/Objectively.vs15/Objectively.vcxproj.filters b/Objectively.vs15/Objectively.vcxproj.filters index 4f52710..3191645 100644 --- a/Objectively.vs15/Objectively.vcxproj.filters +++ b/Objectively.vs15/Objectively.vcxproj.filters @@ -129,6 +129,9 @@ Sources\Objectively + + Sources\Objectively + Sources\Objectively @@ -296,6 +299,9 @@ Sources\Objectively + + Sources\Objectively + Sources\Objectively diff --git a/Objectively.xcodeproj/project.pbxproj b/Objectively.xcodeproj/project.pbxproj index dd65d6d..80754ea 100644 --- a/Objectively.xcodeproj/project.pbxproj +++ b/Objectively.xcodeproj/project.pbxproj @@ -38,6 +38,7 @@ CE4A53601F40E0C800927421 /* PBXTargetDependency */, CE4A53621F40E0C800927421 /* PBXTargetDependency */, CE7F044C2FE05A8B00AE0D63 /* PBXTargetDependency */, + 3027F959E6443E23C73A711B /* PBXTargetDependency */, CE4A53641F40E0C800927421 /* PBXTargetDependency */, CE161E4823BE31780023AC8D /* PBXTargetDependency */, ); @@ -150,10 +151,14 @@ CE76DA2D1C4860130096DD31 /* Objectively.h in Headers */ = {isa = PBXBuildFile; fileRef = CE76D8FC1C481C4E0096DD31 /* Objectively.h */; settings = {ATTRIBUTES = (Public, ); }; }; CE7F04242FD993C600AE0D63 /* URLCachedResponse.h in Headers */ = {isa = PBXBuildFile; fileRef = CE7F04202FD993C600AE0D63 /* URLCachedResponse.h */; settings = {ATTRIBUTES = (Public, ); }; }; CE7F04252FD993C600AE0D63 /* URLCache.h in Headers */ = {isa = PBXBuildFile; fileRef = CE7F04222FD993C600AE0D63 /* URLCache.h */; settings = {ATTRIBUTES = (Public, ); }; }; + C63981E92917506BA5025BBF /* ThreadPool.h in Headers */ = {isa = PBXBuildFile; fileRef = 782FF50CC78357133DAFBF81 /* ThreadPool.h */; settings = {ATTRIBUTES = (Public, ); }; }; CE7F04262FD993C600AE0D63 /* URLCache.c in Sources */ = {isa = PBXBuildFile; fileRef = CE7F04232FD993C600AE0D63 /* URLCache.c */; }; + 266F7D92119FCF6C9B852B75 /* ThreadPool.c in Sources */ = {isa = PBXBuildFile; fileRef = 9C2FE7612EDDB2F063109328 /* ThreadPool.c */; }; CE7F04272FD993C600AE0D63 /* URLCachedResponse.c in Sources */ = {isa = PBXBuildFile; fileRef = CE7F04212FD993C600AE0D63 /* URLCachedResponse.c */; }; CE7F042E2FD9943600AE0D63 /* Objectively.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = CE76D9681C48218E0096DD31 /* Objectively.framework */; }; + BEC2F8C826569B4E03E5ED8F /* Objectively.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = CE76D9681C48218E0096DD31 /* Objectively.framework */; }; CE7F04362FD9947500AE0D63 /* URLCache.c in Sources */ = {isa = PBXBuildFile; fileRef = CE7F04352FD9947500AE0D63 /* URLCache.c */; }; + 920C18A744E410FD9786DEA4 /* ThreadPool.c in Sources */ = {isa = PBXBuildFile; fileRef = 65857D7B78E1D02C982979CE /* ThreadPool.c */; }; CE7F04392FDF2B1100AE0D63 /* JSONContext.h in Headers */ = {isa = PBXBuildFile; fileRef = CE7F04372FDF2B1100AE0D63 /* JSONContext.h */; settings = {ATTRIBUTES = (Public, ); }; }; CE7F043A2FDF2B1100AE0D63 /* JSONContext.c in Sources */ = {isa = PBXBuildFile; fileRef = CE7F04382FDF2B1100AE0D63 /* JSONContext.c */; }; CE7F043B2FDF2B1100AE0D63 /* JSONSerializers.h in Headers */ = {isa = PBXBuildFile; fileRef = CE7F043C2FDF2B1100AE0D63 /* JSONSerializers.h */; settings = {ATTRIBUTES = (Public, ); }; }; @@ -428,6 +433,13 @@ remoteGlobalIDString = CE76D9671C48218E0096DD31; remoteInfo = Objectively; }; + 43D69B38AD06ECEA6CDEA0B5 /* PBXContainerItemProxy */ = { + isa = PBXContainerItemProxy; + containerPortal = CE76D7F31C481B530096DD31 /* Project object */; + proxyType = 1; + remoteGlobalIDString = CE76D9671C48218E0096DD31; + remoteInfo = Objectively; + }; CE7F043D2FE058E700AE0D63 /* PBXContainerItemProxy */ = { isa = PBXContainerItemProxy; containerPortal = CE76D7F31C481B530096DD31 /* Project object */; @@ -449,6 +461,13 @@ remoteGlobalIDString = CE7F04282FD9943600AE0D63; remoteInfo = "Objectively-URLCache"; }; + 7299628DF38C1132501F2E48 /* PBXContainerItemProxy */ = { + isa = PBXContainerItemProxy; + containerPortal = CE76D7F31C481B530096DD31 /* Project object */; + proxyType = 1; + remoteGlobalIDString = 6997E9352EE943DDECA86642; + remoteInfo = "Objectively-ThreadPool"; + }; CE84A8A51DA15B88008BC685 /* PBXContainerItemProxy */ = { isa = PBXContainerItemProxy; containerPortal = CE76D7F31C481B530096DD31 /* Project object */; @@ -674,6 +693,15 @@ ); runOnlyForDeploymentPostprocessing = 1; }; + F756E2B8198B8690009A27C3 /* CopyFiles */ = { + isa = PBXCopyFilesBuildPhase; + buildActionMask = 2147483647; + dstPath = /usr/share/man/man1/; + dstSubfolderSpec = 0; + files = ( + ); + runOnlyForDeploymentPostprocessing = 1; + }; CE7F04422FE058E700AE0D63 /* CopyFiles */ = { isa = PBXCopyFilesBuildPhase; buildActionMask = 2147483647; @@ -1010,9 +1038,13 @@ CE7F04202FD993C600AE0D63 /* URLCachedResponse.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = URLCachedResponse.h; sourceTree = ""; }; CE7F04212FD993C600AE0D63 /* URLCachedResponse.c */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.c; path = URLCachedResponse.c; sourceTree = ""; }; CE7F04222FD993C600AE0D63 /* URLCache.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = URLCache.h; sourceTree = ""; }; + 782FF50CC78357133DAFBF81 /* ThreadPool.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = ThreadPool.h; sourceTree = ""; }; CE7F04232FD993C600AE0D63 /* URLCache.c */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.c; path = URLCache.c; sourceTree = ""; }; + 9C2FE7612EDDB2F063109328 /* ThreadPool.c */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.c; path = ThreadPool.c; sourceTree = ""; }; CE7F04342FD9943600AE0D63 /* Objectively-URLCache */ = {isa = PBXFileReference; explicitFileType = "compiled.mach-o.executable"; includeInIndex = 0; path = "Objectively-URLCache"; sourceTree = BUILT_PRODUCTS_DIR; }; + B9F2267A09C1C219F8FE9A42 /* Objectively-ThreadPool */ = {isa = PBXFileReference; explicitFileType = "compiled.mach-o.executable"; includeInIndex = 0; path = "Objectively-ThreadPool"; sourceTree = BUILT_PRODUCTS_DIR; }; CE7F04352FD9947500AE0D63 /* URLCache.c */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.c; path = URLCache.c; sourceTree = ""; }; + 65857D7B78E1D02C982979CE /* ThreadPool.c */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.c; path = ThreadPool.c; sourceTree = ""; }; CE7F04372FDF2B1100AE0D63 /* JSONContext.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = JSONContext.h; sourceTree = ""; }; CE7F04382FDF2B1100AE0D63 /* JSONContext.c */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.c; path = JSONContext.c; sourceTree = ""; }; CE7F043C2FDF2B1100AE0D63 /* JSONSerializers.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = JSONSerializers.h; sourceTree = ""; }; @@ -1123,6 +1155,14 @@ ); runOnlyForDeploymentPostprocessing = 0; }; + 7F9F7505BEBFC4370394ED0C /* Frameworks */ = { + isa = PBXFrameworksBuildPhase; + buildActionMask = 2147483647; + files = ( + BEC2F8C826569B4E03E5ED8F /* Objectively.framework in Frameworks */, + ); + runOnlyForDeploymentPostprocessing = 0; + }; CE7F04402FE058E700AE0D63 /* Frameworks */ = { isa = PBXFrameworksBuildPhase; buildActionMask = 2147483647; @@ -1415,6 +1455,8 @@ CE594BD21F47BA07004D74FF /* StringReader.h */, CE76D8E91C481C4E0096DD31 /* Thread.c */, CE76D8EA1C481C4E0096DD31 /* Thread.h */, + 9C2FE7612EDDB2F063109328 /* ThreadPool.c */, + 782FF50CC78357133DAFBF81 /* ThreadPool.h */, CE76D8EB1C481C4E0096DD31 /* Types.h */, CE76D8EC1C481C4E0096DD31 /* URL.c */, CE76D8ED1C481C4E0096DD31 /* URL.h */, @@ -1502,6 +1544,7 @@ CE76D9601C481E390096DD31 /* String.c */, CE594BD51F49F8DB004D74FF /* StringReader.c */, CE76D9611C481E390096DD31 /* Thread.c */, + 65857D7B78E1D02C982979CE /* ThreadPool.c */, CE76D9621C481E390096DD31 /* URL.c */, CE7F04352FD9947500AE0D63 /* URLCache.c */, CE76D9631C481E390096DD31 /* URLSession.c */, @@ -1540,6 +1583,7 @@ CEEB02C61F40DD27004C2EDD /* Objectively-Set */, CEEB02D31F40DD29004C2EDD /* Objectively-String */, CEEB02E01F40DD2B004C2EDD /* Objectively-Thread */, + B9F2267A09C1C219F8FE9A42 /* Objectively-ThreadPool */, CEEB02ED1F40DD4E004C2EDD /* Objectively-URL */, CEEB02FA1F40DD4F004C2EDD /* Objectively-URLSession */, CE4A53131F40DFE800927421 /* Objectively-Hello */, @@ -1646,6 +1690,7 @@ CE76DA221C4860130096DD31 /* String.h in Headers */, CE594BD41F47BA07004D74FF /* StringReader.h in Headers */, CE76DA231C4860130096DD31 /* Thread.h in Headers */, + C63981E92917506BA5025BBF /* ThreadPool.h in Headers */, CE76DA241C4860130096DD31 /* Types.h in Headers */, CE76DA251C4860130096DD31 /* URL.h in Headers */, CE7F04252FD993C600AE0D63 /* URLCache.h in Headers */, @@ -1792,6 +1837,24 @@ productReference = CE7F04342FD9943600AE0D63 /* Objectively-URLCache */; productType = "com.apple.product-type.tool"; }; + 6997E9352EE943DDECA86642 /* Objectively-ThreadPool */ = { + isa = PBXNativeTarget; + buildConfigurationList = 4A635B4520A8F0CA5664FB71 /* Build configuration list for PBXNativeTarget "Objectively-ThreadPool" */; + buildPhases = ( + 4F971EB776C08D728076D4C6 /* Sources */, + 7F9F7505BEBFC4370394ED0C /* Frameworks */, + F756E2B8198B8690009A27C3 /* CopyFiles */, + ); + buildRules = ( + ); + dependencies = ( + 703E652D8FC69C337A23B18A /* PBXTargetDependency */, + ); + name = "Objectively-ThreadPool"; + productName = "Objectively-Test"; + productReference = B9F2267A09C1C219F8FE9A42 /* Objectively-ThreadPool */; + productType = "com.apple.product-type.tool"; + }; CE7F043B2FE058E700AE0D63 /* Objectively-RESTClient */ = { isa = PBXNativeTarget; buildConfigurationList = CE7F04432FE058E700AE0D63 /* Build configuration list for PBXNativeTarget "Objectively-RESTClient" */; @@ -2264,6 +2327,7 @@ CEEB02C71F40DD29004C2EDD /* Objectively-String */, CE594BD61F49F931004D74FF /* Objectively-StringReader */, CEEB02D41F40DD2B004C2EDD /* Objectively-Thread */, + 6997E9352EE943DDECA86642 /* Objectively-ThreadPool */, CEEB02E11F40DD4E004C2EDD /* Objectively-URL */, CE7F04282FD9943600AE0D63 /* Objectively-URLCache */, CEEB02EE1F40DD4F004C2EDD /* Objectively-URLSession */, @@ -2304,7 +2368,7 @@ ); runOnlyForDeploymentPostprocessing = 0; shellPath = /bin/sh; - shellScript = "cd $BUILT_PRODUCTS_DIR &&\n./Objectively-Array &&\n./Objectively-Boole &&\n./Objectively-Data &&\n./Objectively-Date &&\n./Objectively-Dictionary &&\n./Objectively-HashTable &&\n./Objectively-IndexPath &&\n./Objectively-IndexSet &&\n./Objectively-JSON \"$SRCROOT/Tests/Objectively/Fixtures/test.json\" &&\n./Objectively-List &&\n./Objectively-Log &&\n./Objectively-Null &&\n./Objectively-Number &&\n./Objectively-Object &&\n./Objectively-Operation &&\n./Objectively-Regexp &&\n./Objectively-Resource &&\n./Objectively-RESTClient &&\n./Objectively-Set &&\n./Objectively-String &&\n./Objectively-StringReader &&\n./Objectively-Thread &&\n./Objectively-URL &&\n./Objectively-URLCache &&\n./Objectively-URLSession &&\n./Objectively-Vector\n"; + shellScript = "cd $BUILT_PRODUCTS_DIR &&\n./Objectively-Array &&\n./Objectively-Boole &&\n./Objectively-Data &&\n./Objectively-Date &&\n./Objectively-Dictionary &&\n./Objectively-HashTable &&\n./Objectively-IndexPath &&\n./Objectively-IndexSet &&\n./Objectively-JSON \"$SRCROOT/Tests/Objectively/Fixtures/test.json\" &&\n./Objectively-List &&\n./Objectively-Log &&\n./Objectively-Null &&\n./Objectively-Number &&\n./Objectively-Object &&\n./Objectively-Operation &&\n./Objectively-Regexp &&\n./Objectively-Resource &&\n./Objectively-RESTClient &&\n./Objectively-Set &&\n./Objectively-String &&\n./Objectively-StringReader &&\n./Objectively-Thread &&\n./Objectively-ThreadPool &&\n./Objectively-URL &&\n./Objectively-URLCache &&\n./Objectively-URLSession &&\n./Objectively-Vector\n"; }; CEC0F1600000000000000001 /* Configure */ = { isa = PBXShellScriptBuildPhase; @@ -2403,6 +2467,7 @@ CE76D98A1C4821CE0096DD31 /* String.c in Sources */, CE594BD31F47BA07004D74FF /* StringReader.c in Sources */, CE76D98B1C4821CE0096DD31 /* Thread.c in Sources */, + 266F7D92119FCF6C9B852B75 /* ThreadPool.c in Sources */, CE76D98C1C4821CE0096DD31 /* URL.c in Sources */, CE7F04262FD993C600AE0D63 /* URLCache.c in Sources */, CE7F04272FD993C600AE0D63 /* URLCachedResponse.c in Sources */, @@ -2426,6 +2491,14 @@ ); runOnlyForDeploymentPostprocessing = 0; }; + 4F971EB776C08D728076D4C6 /* Sources */ = { + isa = PBXSourcesBuildPhase; + buildActionMask = 2147483647; + files = ( + 920C18A744E410FD9786DEA4 /* ThreadPool.c in Sources */, + ); + runOnlyForDeploymentPostprocessing = 0; + }; CE7F043E2FE058E700AE0D63 /* Sources */ = { isa = PBXSourcesBuildPhase; buildActionMask = 2147483647; @@ -2763,6 +2836,11 @@ target = CE76D9671C48218E0096DD31 /* Objectively */; targetProxy = CE7F042A2FD9943600AE0D63 /* PBXContainerItemProxy */; }; + 703E652D8FC69C337A23B18A /* PBXTargetDependency */ = { + isa = PBXTargetDependency; + target = CE76D9671C48218E0096DD31 /* Objectively */; + targetProxy = 43D69B38AD06ECEA6CDEA0B5 /* PBXContainerItemProxy */; + }; CE7F043C2FE058E700AE0D63 /* PBXTargetDependency */ = { isa = PBXTargetDependency; target = CE76D9671C48218E0096DD31 /* Objectively */; @@ -2778,6 +2856,11 @@ target = CE7F04282FD9943600AE0D63 /* Objectively-URLCache */; targetProxy = CE7F044B2FE05A8B00AE0D63 /* PBXContainerItemProxy */; }; + 3027F959E6443E23C73A711B /* PBXTargetDependency */ = { + isa = PBXTargetDependency; + target = 6997E9352EE943DDECA86642 /* Objectively-ThreadPool */; + targetProxy = 7299628DF38C1132501F2E48 /* PBXContainerItemProxy */; + }; CE84A8A61DA15B88008BC685 /* PBXTargetDependency */ = { isa = PBXTargetDependency; target = CE76D9671C48218E0096DD31 /* Objectively */; @@ -3151,6 +3234,15 @@ }; name = Debug; }; + C2A2B533960082B9A5EB65FC /* Debug */ = { + isa = XCBuildConfiguration; + buildSettings = { + LIBRARY_SEARCH_PATHS = "$(HOMEBREW_PREFIX)/lib"; + OTHER_LDFLAGS = "-lcheck"; + PRODUCT_NAME = "$(TARGET_NAME)"; + }; + name = Debug; + }; CE7F04332FD9943600AE0D63 /* Release */ = { isa = XCBuildConfiguration; buildSettings = { @@ -3160,6 +3252,15 @@ }; name = Release; }; + 19A7AC8AA20D613DDD05D0D7 /* Release */ = { + isa = XCBuildConfiguration; + buildSettings = { + LIBRARY_SEARCH_PATHS = "$(HOMEBREW_PREFIX)/lib"; + OTHER_LDFLAGS = "-lcheck"; + PRODUCT_NAME = "$(TARGET_NAME)"; + }; + name = Release; + }; CE7F04442FE058E700AE0D63 /* Debug */ = { isa = XCBuildConfiguration; buildSettings = { @@ -3667,6 +3768,15 @@ defaultConfigurationIsVisible = 0; defaultConfigurationName = Release; }; + 4A635B4520A8F0CA5664FB71 /* Build configuration list for PBXNativeTarget "Objectively-ThreadPool" */ = { + isa = XCConfigurationList; + buildConfigurations = ( + C2A2B533960082B9A5EB65FC /* Debug */, + 19A7AC8AA20D613DDD05D0D7 /* Release */, + ); + defaultConfigurationIsVisible = 0; + defaultConfigurationName = Release; + }; CE7F04432FE058E700AE0D63 /* Build configuration list for PBXNativeTarget "Objectively-RESTClient" */ = { isa = XCConfigurationList; buildConfigurations = ( diff --git a/Objectively.xcodeproj/xcshareddata/xcschemes/Objectively-ThreadPool.xcscheme b/Objectively.xcodeproj/xcshareddata/xcschemes/Objectively-ThreadPool.xcscheme new file mode 100644 index 0000000..d091e9f --- /dev/null +++ b/Objectively.xcodeproj/xcshareddata/xcschemes/Objectively-ThreadPool.xcscheme @@ -0,0 +1,80 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/README.md b/README.md index 85d1d99..e2fdfed 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Objectively is a cross-platform framework that provides rich object-oriented sem * **Automatic memory management** with reference counting * **Unicode strings** with mutlibyte character support * **Collections** for Objects and C types: Array, Dictionary, List, Set and more - * **Concurrency**: Lock, Condition, Thread, Operation, OperationQueue + * **Concurrency**: Lock, Condition, Thread, ThreadPool, Operation, OperationQueue * **Networking**: URLSession, JSONContext & RESTClient ## tl;dr diff --git a/Sources/Objectively.h b/Sources/Objectively.h index 5021e38..5eba8ec 100644 --- a/Sources/Objectively.h +++ b/Sources/Objectively.h @@ -65,6 +65,7 @@ #include #include #include +#include #include #include #include diff --git a/Sources/Objectively/Makefile.am b/Sources/Objectively/Makefile.am index bf57931..8bb2588 100644 --- a/Sources/Objectively/Makefile.am +++ b/Sources/Objectively/Makefile.am @@ -43,6 +43,7 @@ pkginclude_HEADERS = \ String.h \ StringReader.h \ Thread.h \ + ThreadPool.h \ Types.h \ URL.h \ URLCache.h \ @@ -96,6 +97,7 @@ libObjectively_la_SOURCES = \ String.c \ StringReader.c \ Thread.c \ + ThreadPool.c \ URL.c \ URLCache.c \ URLCachedResponse.c \ diff --git a/Sources/Objectively/Operation.c b/Sources/Objectively/Operation.c index 13eb17f..402fc5b 100644 --- a/Sources/Objectively/Operation.c +++ b/Sources/Objectively/Operation.c @@ -44,8 +44,8 @@ static void dealloc(Object *self) { Operation *this = (Operation *) self; - release(this->locals.condition); - release(this->locals.dependencies); + release(this->condition); + release(this->dependencies); super(Object, self, dealloc); } @@ -61,9 +61,9 @@ static void addDependency(Operation *self, Operation *dependency) { assert(dependency); assert(dependency != self); - assert($((Array *) self->locals.dependencies, indexOfObject, dependency) == -1); + assert($((Array *) self->dependencies, indexOfObject, dependency) == -1); - $(self->locals.dependencies, addObject, dependency); + $(self->dependencies, addObject, dependency); } /** @@ -72,18 +72,18 @@ static void addDependency(Operation *self, Operation *dependency) { */ static void cancel(Operation *self) { - if (self->isCancelled == false) { - if (self->isFinished == false) { - if (self->isExecuting == false) { - self->isCancelled = true; - - OperationQueue *queue = self->locals.queue; - if (queue) { - synchronized(queue->locals.condition, { - $(queue->locals.condition, signal); - }); - } - } + if (__atomic_load_n(&self->isCancelled, __ATOMIC_ACQUIRE) || self->isFinished) { + return; + } + + __atomic_store_n(&self->isCancelled, true, __ATOMIC_RELEASE); + + if (self->isExecuting == false) { + OperationQueue *queue = self->queue; + if (queue) { + synchronized(queue->condition, { + $(queue->condition, signal); + }); } } } @@ -94,7 +94,7 @@ static void cancel(Operation *self) { */ static Array *dependencies(const Operation *self) { - ident dependencies = $((Object *) self->locals.dependencies, copy); + ident dependencies = $((Object *) self->dependencies, copy); return (Array *) dependencies; } @@ -108,11 +108,11 @@ static Operation *init(Operation *self) { self = (Operation *) super(Object, self, init); if (self) { - self->locals.condition = $(alloc(Condition), init); - assert(self->locals.condition); + self->condition = $(alloc(Condition), init); + assert(self->condition); - self->locals.dependencies = $(alloc(Array), init); - assert(self->locals.dependencies); + self->dependencies = $(alloc(Array), init); + assert(self->dependencies); } return self; @@ -143,11 +143,11 @@ static bool isReady(const Operation *self) { return false; } - if (self->isCancelled) { + if (__atomic_load_n(&self->isCancelled, __ATOMIC_ACQUIRE)) { return true; } - const Array *dependencies = (Array *) self->locals.dependencies; + const Array *dependencies = (Array *) self->dependencies; for (size_t i = 0; i < dependencies->count; i++) { Operation *dependency = $(dependencies, objectAtIndex, i); @@ -167,7 +167,7 @@ static void removeDepdenency(Operation *self, Operation *dependency) { assert(dependency); - $(self->locals.dependencies, removeObject, dependency); + $(self->dependencies, removeObject, dependency); } /** @@ -180,7 +180,7 @@ static void start(Operation *self) { return; } - if (self->isCancelled == false) { + if (__atomic_load_n(&self->isCancelled, __ATOMIC_ACQUIRE) == false) { self->isExecuting = true; self->function(self); self->isExecuting = false; @@ -188,8 +188,8 @@ static void start(Operation *self) { self->isFinished = true; - synchronized(self->locals.condition, { - $(self->locals.condition, broadcast); + synchronized(self->condition, { + $(self->condition, broadcast); }); } @@ -199,9 +199,9 @@ static void start(Operation *self) { */ static void waitUntilFinished(const Operation *self) { - synchronized(self->locals.condition, { + synchronized(self->condition, { while (self->isFinished == false) { - $(self->locals.condition, wait); + $(self->condition, wait); } }); } diff --git a/Sources/Objectively/Operation.h b/Sources/Objectively/Operation.h index 51d6b7d..66224d6 100644 --- a/Sources/Objectively/Operation.h +++ b/Sources/Objectively/Operation.h @@ -63,40 +63,29 @@ struct Operation { */ OperationInterface *interface[0]; - /** - * @private - */ - struct { - /** - * @brief The Condition enabling `waitUntilFinished`. - */ - Condition *condition; - - /** - * @brief Contains Operations which must finish before this one can start. - */ - Array *dependencies; - - /** - * @brief The OperationQueue this Operation was added to, if any. - * @remarks Not retained; an OperationQueue retains its Operations. This - * exists so that `cancel` can notify the queue, which must re-evaluate - * because cancelling an Operation makes it ready. - */ - OperationQueue *queue; - } locals; - /** * @brief If `true`, this Operation will be expected to coordinate its own * concurrency and internal state management by overriding `start`. */ bool asynchronous; + /** + * @brief The Condition enabling `waitUntilFinished`. + * @private + */ + Condition *condition; + /** * @brief The user data. */ ident data; + /** + * @brief Contains Operations which must finish before this one can start. + * @private + */ + Array *dependencies; + /** * @brief The Operation function. */ @@ -104,9 +93,11 @@ struct Operation { /** * @brief `true` when this Operation has been cancelled, `false` otherwise. + * @remarks Operation::cancel MAY set this from another thread while the Operation executes, so + * the function SHOULD read it with `__atomic_load_n(&operation->isCancelled, __ATOMIC_ACQUIRE)`. */ bool isCancelled; - + /** * @brief True once an OperationQueue has dispatched this Operation to one of * its Threads, so that no other Thread also starts it. @@ -122,6 +113,15 @@ struct Operation { * @brief `true` when this Operation is finished, `false` otherwise. */ bool isFinished; + + /** + * @brief The OperationQueue this Operation was added to, if any. + * @remarks Not retained; an OperationQueue retains its Operations. This + * exists so that `cancel` can notify the queue, which must re-evaluate + * because cancelling an Operation makes it ready. + * @private + */ + OperationQueue *queue; }; /** @@ -147,6 +147,8 @@ struct OperationInterface { * @fn void Operation::cancel(Operation *self) * @brief Cancels this Operation, allowing it to complete immediately. * @param self The Operation. + * @remarks An Operation that has not started will not run its function. An Operation that is + * executing runs to completion, but its function MAY check `isCancelled` and return early. * @memberof Operation */ void (*cancel)(Operation *self); diff --git a/Sources/Objectively/OperationQueue.c b/Sources/Objectively/OperationQueue.c index 970231d..6576d98 100644 --- a/Sources/Objectively/OperationQueue.c +++ b/Sources/Objectively/OperationQueue.c @@ -44,23 +44,14 @@ static void dealloc(Object *self) { OperationQueue *this = (OperationQueue *) self; - const Array *threads = (Array *) this->locals.threads; - - for (size_t i = 0; i < threads->count; i++) { - $((Thread *) $(threads, objectAtIndex, i), cancel); - } - - synchronized(this->locals.condition, { - $(this->locals.condition, broadcast); + synchronized(this->condition, { + this->isCancelled = true; + $(this->condition, broadcast); }); - for (size_t i = 0; i < threads->count; i++) { - $((Thread *) $(threads, objectAtIndex, i), join, NULL); - } - - release(this->locals.threads); - release(this->locals.condition); - release(this->locals.operations); + release(this->threadPool); + release(this->condition); + release(this->operations); super(Object, self, dealloc); } @@ -78,10 +69,10 @@ static void addOperation(OperationQueue *self, Operation *operation) { assert(operation->isExecuting == false); assert(operation->isFinished == false); - synchronized(self->locals.condition, { - operation->locals.queue = self; - $(self->locals.operations, addObject, operation); - $(self->locals.condition, signal); + synchronized(self->condition, { + operation->queue = self; + $(self->operations, addObject, operation); + $(self->condition, signal); }); } @@ -145,21 +136,21 @@ static ident run(Thread *thread) { Operation *operation = NULL; - synchronized(self->locals.condition, { + synchronized(self->condition, { - isCancelled = thread->isCancelled; + isCancelled = self->isCancelled; if (!isCancelled && self->isSuspended == false) { - operation = $(self->locals.operations, find, isOperationReady, NULL); + operation = $(self->operations, find, isOperationReady, NULL); if (operation) { operation->isDispatched = true; - $(self->locals.condition, signal); + $(self->condition, signal); } } if (operation == NULL && !isCancelled) { - $(self->locals.condition, wait); + $(self->condition, wait); } }); @@ -194,28 +185,17 @@ static OperationQueue *initWithMaxConcurrentOperations(OperationQueue *self, siz self = (OperationQueue *) super(Object, self, init); if (self) { - self->locals.condition = $(alloc(Condition), init); - assert(self->locals.condition); + self->condition = $(alloc(Condition), init); + assert(self->condition); - self->locals.operations = $(alloc(Array), init); - assert(self->locals.operations); + self->operations = $(alloc(Array), init); + assert(self->operations); - self->locals.threads = $(alloc(Array), init); - assert(self->locals.threads); + self->threadPool = $(alloc(ThreadPool), initWithSize, maxConcurrentOperations); + assert(self->threadPool); for (size_t i = 0; i < maxConcurrentOperations; i++) { - - Thread *thread = $(alloc(Thread), initWithFunction, run, self); - assert(thread); - - $(self->locals.threads, addObject, thread); - - release(thread); - } - - const Array *threads = (Array *) self->locals.threads; - for (size_t i = 0; i < threads->count; i++) { - $((Thread *) $(threads, objectAtIndex, i), start); + $(self->threadPool, detach, run, self); } } @@ -230,8 +210,8 @@ static size_t operationCount(const OperationQueue *self) { size_t count; - synchronized(self->locals.condition, { - count = ((Array *) self->locals.operations)->count; + synchronized(self->condition, { + count = ((Array *) self->operations)->count; }); return count; @@ -245,8 +225,8 @@ static Array *operations(const OperationQueue *self) { ident operations; - synchronized(self->locals.condition, { - operations = $((Object * ) self->locals.operations, copy); + synchronized(self->condition, { + operations = $((Object * ) self->operations, copy); }); return (Array *) operations; @@ -261,10 +241,10 @@ static void removeOperation(OperationQueue *self, Operation *operation) { assert(operation); assert(operation->isExecuting == false); - synchronized(self->locals.condition, { - operation->locals.queue = NULL; - $(self->locals.operations, removeObject, operation); - $(self->locals.condition, signal); + synchronized(self->condition, { + operation->queue = NULL; + $(self->operations, removeObject, operation); + $(self->condition, signal); }); } @@ -274,9 +254,9 @@ static void removeOperation(OperationQueue *self, Operation *operation) { */ static void resume(OperationQueue *self) { - synchronized(self->locals.condition, { + synchronized(self->condition, { self->isSuspended = false; - $(self->locals.condition, broadcast); + $(self->condition, broadcast); }); } @@ -286,7 +266,7 @@ static void resume(OperationQueue *self) { */ static void suspend(OperationQueue *self) { - synchronized(self->locals.condition, { + synchronized(self->condition, { self->isSuspended = true; }); } @@ -301,8 +281,8 @@ static void waitUntilAllOperationsAreFinished(OperationQueue *self) { Operation *operation = NULL; - synchronized(self->locals.condition, { - const Array *operations = (Array *) self->locals.operations; + synchronized(self->condition, { + const Array *operations = (Array *) self->operations; if (operations->count) { operation = retain($(operations, firstObject)); } diff --git a/Sources/Objectively/OperationQueue.h b/Sources/Objectively/OperationQueue.h index 47bdcd1..147537d 100644 --- a/Sources/Objectively/OperationQueue.h +++ b/Sources/Objectively/OperationQueue.h @@ -26,7 +26,7 @@ #include #include #include -#include +#include /** * @file @@ -55,26 +55,16 @@ struct OperationQueue { OperationQueueInterface *interface[0]; /** + * @brief A condition signaled on `addOperation` and `removeOperation`. * @private */ - struct { + Condition *condition; - /** - * @brief A condition signaled on `addOperation` and `removeOperation`. - */ - Condition *condition; - - /** - * @brief The Operations. - */ - Array *operations; - - /** - * @brief The backing Threads, one per concurrently executing Operation. - */ - Array *threads; - - } locals; + /** + * @brief `true` when this queue is being deallocated, which ends its ThreadFunctions. + * @private + */ + bool isCancelled; /** * @brief When `true`, the queue will not `start` any new Operations. @@ -82,6 +72,18 @@ struct OperationQueue { * Threads of the transition. */ bool isSuspended; + + /** + * @brief The Operations. + * @private + */ + Array *operations; + + /** + * @brief The backing ThreadPool, with one Thread per concurrently executing Operation. + * @private + */ + ThreadPool *threadPool; }; /** diff --git a/Sources/Objectively/RESTClient.c b/Sources/Objectively/RESTClient.c index 24b094b..6747dab 100644 --- a/Sources/Objectively/RESTClient.c +++ b/Sources/Objectively/RESTClient.c @@ -24,6 +24,8 @@ #include #include +#include + #include "RESTClient.h" #include "URL.h" #include "URLRequest.h" @@ -42,10 +44,13 @@ static void RESTClient_AsyncCompletion(URLSessionTask *task, bool success) { RESTClient_AsyncState *state = task->data; URLSessionDataTask *dataTask = (URLSessionDataTask *) task; - const int status = task->response ? task->response->httpStatusCode : 0; if (state->completion) { - state->completion(status, dataTask->data, state->userData); + if (success) { + state->completion(task->response->httpStatusCode, dataTask->data, state->userData); + } else { + state->completion(0, NULL, state->userData); + } } free(state); @@ -85,10 +90,17 @@ static int request(RESTClient *self, HTTPMethod method, const char *url_string, release(request); } + if (task == NULL) { + return 0; + } + $((URLSessionTask *) task, execute); - const int status = task->urlSessionTask.response->httpStatusCode; - if (out_data && task->data) { + const URLSessionTask *t = (URLSessionTask *) task; + const bool success = t->state == URLSESSIONTASK_COMPLETED && t->result == CURLE_OK; + + const int status = success ? t->response->httpStatusCode : 0; + if (success && out_data && task->data) { *out_data = retain(task->data); } @@ -123,6 +135,13 @@ static void requestAsync(RESTClient *self, HTTPMethod method, const char *url_st release(request); } + if (task == NULL) { + if (completion) { + completion(0, NULL, userData); + } + return; + } + RESTClient_AsyncState *state = calloc(1, sizeof(RESTClient_AsyncState)); state->completion = completion; state->userData = userData; diff --git a/Sources/Objectively/RESTClient.h b/Sources/Objectively/RESTClient.h index 90aee30..51584bb 100644 --- a/Sources/Objectively/RESTClient.h +++ b/Sources/Objectively/RESTClient.h @@ -42,7 +42,7 @@ typedef struct RESTClientInterface RESTClientInterface; /** * @brief A completion handler for asynchronous RESTClient requests. - * @param status The HTTP response status code, or `0` on connection failure. + * @param status The HTTP response status code, or `0` if the transfer failed or was cancelled. * @param data The response body, or `NULL`. * @param userData The user data pointer passed to the originating async method. * @ingroup REST @@ -96,7 +96,7 @@ struct RESTClientInterface { * @param headers A `NULL`-terminated array of alternating field/value C strings * (e.g. `{ "X-Field", "value", NULL }`) to add to the request, or `NULL` for none. * @param data Optionally receives the retained response body. Caller must `release`. - * @return The HTTP response status code, or `0` on connection failure. + * @return The HTTP response status code, or `0` if the transfer failed or was cancelled. * @memberof RESTClient */ int (*del)(RESTClient *self, const char *url, const char **headers, Data **data); @@ -123,7 +123,7 @@ struct RESTClientInterface { * @param headers A `NULL`-terminated array of alternating field/value C strings * (e.g. `{ "X-Field", "value", NULL }`) to add to the request, or `NULL` for none. * @param data Optionally receives the retained response body. Caller must `release`. - * @return The HTTP response status code, or `0` on connection failure. + * @return The HTTP response status code, or `0` if the transfer failed or was cancelled. * @memberof RESTClient */ int (*get)(RESTClient *self, const char *url, const char **headers, Data **data); @@ -149,7 +149,7 @@ struct RESTClientInterface { * @param url The URL string. * @param headers A `NULL`-terminated array of alternating field/value C strings * (e.g. `{ "X-Field", "value", NULL }`) to add to the request, or `NULL` for none. - * @return The HTTP response status code, or `0` on connection failure. + * @return The HTTP response status code, or `0` if the transfer failed or was cancelled. * @memberof RESTClient */ int (*head)(RESTClient *self, const char *url, const char **headers); @@ -195,7 +195,7 @@ struct RESTClientInterface { * @param headers A `NULL`-terminated array of alternating field/value C strings * (e.g. `{ "X-Field", "value", NULL }`) to add to the request, or `NULL` for none. * @param data Optionally receives the retained response body. Caller must `release`. - * @return The HTTP response status code, or `0` on connection failure. + * @return The HTTP response status code, or `0` if the transfer failed or was cancelled. * @memberof RESTClient */ int (*options)(RESTClient *self, const char *url, const char **headers, Data **data); @@ -223,7 +223,7 @@ struct RESTClientInterface { * @param headers A `NULL`-terminated array of alternating field/value C strings * (e.g. `{ "X-Field", "value", NULL }`) to add to the request, or `NULL` for none. * @param data Optionally receives the retained response body. Caller must `release`. - * @return The HTTP response status code, or `0` on connection failure. + * @return The HTTP response status code, or `0` if the transfer failed or was cancelled. * @memberof RESTClient */ int (*patch)(RESTClient *self, const char *url, const Data *body, const char **headers, Data **data); @@ -252,7 +252,7 @@ struct RESTClientInterface { * @param headers A `NULL`-terminated array of alternating field/value C strings * (e.g. `{ "X-Field", "value", NULL }`) to add to the request, or `NULL` for none. * @param data Optionally receives the retained response body. Caller must `release`. - * @return The HTTP response status code, or `0` on connection failure. + * @return The HTTP response status code, or `0` if the transfer failed or was cancelled. * @memberof RESTClient */ int (*post)(RESTClient *self, const char *url, const Data *body, const char **headers, Data **data); @@ -281,7 +281,7 @@ struct RESTClientInterface { * @param headers A `NULL`-terminated array of alternating field/value C strings * (e.g. `{ "X-Field", "value", NULL }`) to add to the request, or `NULL` for none. * @param data Optionally receives the retained response body. Caller must `release`. - * @return The HTTP response status code, or `0` on connection failure. + * @return The HTTP response status code, or `0` if the transfer failed or was cancelled. * @memberof RESTClient */ int (*put)(RESTClient *self, const char *url, const Data *body, const char **headers, Data **data); diff --git a/Sources/Objectively/Thread.c b/Sources/Objectively/Thread.c index 02c282c..34bd7fb 100644 --- a/Sources/Objectively/Thread.c +++ b/Sources/Objectively/Thread.c @@ -69,7 +69,7 @@ static void cancel(Thread *self) { // int err = pthread_cancel(*((pthread_t *) self->thread)); // assert(err == 0); - self->isCancelled = true; + __atomic_store_n(&self->isCancelled, true, __ATOMIC_RELEASE); } static __thread Thread *_currentThread; diff --git a/Sources/Objectively/Thread.h b/Sources/Objectively/Thread.h index 8a3e341..5ee491c 100644 --- a/Sources/Objectively/Thread.h +++ b/Sources/Objectively/Thread.h @@ -75,6 +75,8 @@ struct Thread { /** * @brief `true` when this Thread has been cancelled, `false` otherwise. + * @remarks Thread::cancel sets this from another thread, so the ThreadFunction SHOULD read it + * with `__atomic_load_n(&thread->isCancelled, __ATOMIC_ACQUIRE)`. */ bool isCancelled; diff --git a/Sources/Objectively/ThreadPool.c b/Sources/Objectively/ThreadPool.c new file mode 100644 index 0000000..b76b736 --- /dev/null +++ b/Sources/Objectively/ThreadPool.c @@ -0,0 +1,392 @@ +/* + * Objectively: Ultra-lightweight object oriented framework for GNU C. + * Copyright (C) 2014 Jay Dolan + * + * This software is provided 'as-is', without any express or implied + * warranty. In no event will the authors be held liable for any damages + * arising from the use of this software. + * + * Permission is granted to anyone to use this software for any purpose, + * including commercial applications, and to alter it and redistribute it + * freely, subject to the following restrictions: + * + * 1. The origin of this software must not be misrepresented; you must not + * claim that you wrote the original software. If you use this software + * in a product, an acknowledgment in the product documentation would be + * appreciated but is not required. + * + * 2. Altered source versions must be plainly marked as such, and must not be + * misrepresented as being the original software. + * + * 3. This notice may not be removed or altered from any source distribution. + */ + +#include "Config.h" + +#include +#include + +#if defined(_WIN32) +#include +#endif + +#if HAVE_UNISTD_H +#include +#endif + +#include "ThreadPool.h" + +#define _Class _ThreadPool + +/** + * @brief The states of a ThreadPoolWork. + */ +typedef enum { + ThreadPoolWorkPending, + ThreadPoolWorkRunning, + ThreadPoolWorkFinished, +} ThreadPoolWorkState; + +struct ThreadPoolWork { + /** + * @brief The ThreadFunction. + */ + ThreadFunction function; + + /** + * @brief The user data. + */ + ident data; + + /** + * @brief `true` if the Thread that runs this ThreadPoolWork frees it. + */ + bool isDetached; + + /** + * @brief The next ThreadPoolWork in the queue, or `NULL`. + */ + ThreadPoolWork *next; + + /** + * @brief The ThreadPool owning this work (weak reference). + */ + ThreadPool *pool; + + /** + * @brief The return value of the ThreadFunction. + */ + ident result; + + /** + * @brief The state. + */ + ThreadPoolWorkState state; +}; + +#pragma mark - Object + +/** + * @see Object::copy(const Object *) + */ +static Object *copy(const Object *self) { + return NULL; +} + +/** + * @see Object::dealloc(Object *) + */ +static void dealloc(Object *self) { + + ThreadPool *this = (ThreadPool *) self; + + const Array *threads = this->threads; + + synchronized(this->condition, { + for (size_t i = 0; i < threads->count; i++) { + $((Thread *) $(threads, objectAtIndex, i), cancel); + } + $(this->condition, broadcast); + }); + + for (size_t i = 0; i < threads->count; i++) { + $((Thread *) $(threads, objectAtIndex, i), join, NULL); + } + + assert(this->head == NULL); + + release(this->threads); + + release(this->condition); + + super(Object, self, dealloc); +} + +#pragma mark - ThreadPool + +/** + * @brief Runs `work` on the calling thread, with a Thread that is never started. + */ +static void runOnCallingThread(ThreadPoolWork *work) { + + Thread *thread = $(alloc(Thread), initWithFunction, work->function, work->data); + assert(thread); + + work->result = work->function(thread); + work->state = ThreadPoolWorkFinished; + + release(thread); +} + +/** + * @brief Queues `function`, or runs it on the calling thread if this ThreadPool has no Threads. + * @return The ThreadPoolWork, or `NULL` if `isDetached`. + */ +static ThreadPoolWork *submit(ThreadPool *self, ThreadFunction function, ident data, bool isDetached) { + + assert(function); + + ThreadPoolWork *work = calloc(1, sizeof(ThreadPoolWork)); + assert(work); + + work->pool = self; + work->state = ThreadPoolWorkPending; + work->isDetached = isDetached; + work->function = function; + work->data = data; + + if (self->size == 0) { + runOnCallingThread(work); + if (isDetached) { + free(work); + return NULL; + } + return work; + } + + synchronized(self->condition, { + if (self->tail) { + self->tail->next = work; + } else { + self->head = work; + } + self->tail = work; + $(self->condition, broadcast); + }); + + return isDetached ? NULL : work; +} + +/** + * @fn void ThreadPool::detach(ThreadPool *self, ThreadFunction function, ident data) + * @memberof ThreadPool + */ +static void detach(ThreadPool *self, ThreadFunction function, ident data) { + submit(self, function, data, true); +} + +/** + * @fn ThreadPoolWork *ThreadPool::execute(ThreadPool *self, ThreadFunction function, ident data) + * @memberof ThreadPool + */ +static ThreadPoolWork *execute(ThreadPool *self, ThreadFunction function, ident data) { + return submit(self, function, data, false); +} + +/** + * @fn ThreadPool *ThreadPool::init(ThreadPool *self) + * @memberof ThreadPool + */ +static ThreadPool *init(ThreadPool *self) { + +#if defined(_WIN32) + const size_t size = GetActiveProcessorCount(ALL_PROCESSOR_GROUPS); +#elif defined(_SC_NPROCESSORS_ONLN) + const long processors = sysconf(_SC_NPROCESSORS_ONLN); + const size_t size = processors > 0 ? (size_t) processors : 1; +#else + const size_t size = 1; +#endif + + return $(self, initWithSize, size); +} + +/** + * @brief Removes `work` from the queue. + * @remarks The caller MUST hold the condition. + */ +static void dequeue(ThreadPool *self, ThreadPoolWork *work) { + + ThreadPoolWork *previous = NULL; + for (ThreadPoolWork *w = self->head; w; previous = w, w = w->next) { + if (w == work) { + if (previous) { + previous->next = work->next; + } else { + self->head = work->next; + } + if (self->tail == work) { + self->tail = previous; + } + work->next = NULL; + return; + } + } + + assert(false); +} + +/** + * @brief ThreadFunction for the ThreadPool Threads. + * @remarks `thread->data` is the user data of the current ThreadPoolWork, so the ThreadPool is + * kept in a local. + */ +static ident run(Thread *thread) { + + ThreadPool *self = thread->data; + Condition *condition = self->condition; + + $((Lock *) condition, lock); + + while (true) { + + while (self->head == NULL && thread->isCancelled == false) { + $(condition, wait); + } + + ThreadPoolWork *work = self->head; + if (work == NULL) { + break; + } + + dequeue(self, work); + work->state = ThreadPoolWorkRunning; + + $((Lock *) condition, unlock); + + thread->data = work->data; + const ident result = work->function(thread); + + $((Lock *) condition, lock); + + if (work->isDetached) { + free(work); + } else { + work->result = result; + work->state = ThreadPoolWorkFinished; + $(condition, broadcast); + } + } + + $((Lock *) condition, unlock); + + return NULL; +} + +/** + * @fn ThreadPool *ThreadPool::initWithSize(ThreadPool *self, size_t size) + * @memberof ThreadPool + */ +static ThreadPool *initWithSize(ThreadPool *self, size_t size) { + + self = (ThreadPool *) super(Object, self, init); + if (self) { + + self->condition = $(alloc(Condition), init); + assert(self->condition); + + self->size = size; + + self->threads = $(alloc(Array), initWithCapacity, size); + assert(self->threads); + + for (size_t i = 0; i < self->size; i++) { + Thread *thread = $(alloc(Thread), initWithFunction, run, self); + assert(thread); + + $(self->threads, addObject, thread); + $(thread, start); + + release(thread); + } + } + + return self; +} + +/** + * @fn ident ThreadPool::wait(ThreadPool *self, ThreadPoolWork *work) + * @memberof ThreadPool + */ +static ident _wait(ThreadPool *self, ThreadPoolWork *work) { + + if (work == NULL) { + return NULL; + } + + assert(work->pool == self); + assert(work->isDetached == false); + + bool isPending = false; + + synchronized(self->condition, { + if (work->state == ThreadPoolWorkPending) { + dequeue(self, work); + work->state = ThreadPoolWorkRunning; + isPending = true; + } else { + while (work->state != ThreadPoolWorkFinished) { + $(self->condition, wait); + } + } + }); + + if (isPending) { + runOnCallingThread(work); + } + + const ident result = work->result; + free(work); + + return result; +} + +#pragma mark - Class lifecycle + +/** + * @see Class::initialize(Class *) + */ +static void initialize(Class *clazz) { + + ((ObjectInterface *) clazz->interface)->copy = copy; + ((ObjectInterface *) clazz->interface)->dealloc = dealloc; + + ((ThreadPoolInterface *) clazz->interface)->detach = detach; + ((ThreadPoolInterface *) clazz->interface)->execute = execute; + ((ThreadPoolInterface *) clazz->interface)->init = init; + ((ThreadPoolInterface *) clazz->interface)->initWithSize = initWithSize; + ((ThreadPoolInterface *) clazz->interface)->wait = _wait; +} + +/** + * @fn Class *ThreadPool::_ThreadPool(void) + * @memberof ThreadPool + */ +Class *_ThreadPool(void) { + static Class *clazz; + static Once once; + + do_once(&once, { + clazz = _initialize(&(const ClassDef) { + .name = "ThreadPool", + .superclass = _Object(), + .instanceSize = sizeof(ThreadPool), + .interfaceSize = sizeof(ThreadPoolInterface), + .initialize = initialize, + }); + }); + + return clazz; +} + +#undef _Class diff --git a/Sources/Objectively/ThreadPool.h b/Sources/Objectively/ThreadPool.h new file mode 100644 index 0000000..d90ae8e --- /dev/null +++ b/Sources/Objectively/ThreadPool.h @@ -0,0 +1,178 @@ +/* + * Objectively: Ultra-lightweight object oriented framework for GNU C. + * Copyright (C) 2014 Jay Dolan + * + * This software is provided 'as-is', without any express or implied + * warranty. In no event will the authors be held liable for any damages + * arising from the use of this software. + * + * Permission is granted to anyone to use this software for any purpose, + * including commercial applications, and to alter it and redistribute it + * freely, subject to the following restrictions: + * + * 1. The origin of this software must not be misrepresented; you must not + * claim that you wrote the original software. If you use this software + * in a product, an acknowledgment in the product documentation would be + * appreciated but is not required. + * + * 2. Altered source versions must be plainly marked as such, and must not be + * misrepresented as being the original software. + * + * 3. This notice may not be removed or altered from any source distribution. + */ + +#pragma once + +#include +#include +#include +#include + +/** + * @file + * @brief A fixed set of Threads that run ThreadFunctions on behalf of the caller. + */ + +typedef struct ThreadPool ThreadPool; +typedef struct ThreadPoolInterface ThreadPoolInterface; + +/** + * @brief An opaque handle to a ThreadFunction submitted with ThreadPool::execute. + * @remarks A ThreadPoolWork MUST be passed to ThreadPool::wait exactly once, before the ThreadPool + * is released. ThreadPool::wait frees it. + */ +typedef struct ThreadPoolWork ThreadPoolWork; + +/** + * @brief A fixed set of Threads that run ThreadFunctions on behalf of the caller. + * @details Submitted functions wait in a first-in, first-out queue until a Thread is free. Each + * ThreadFunction receives a Thread whose `data` is the user data that was submitted with the + * function. + * + * Deallocating a ThreadPool sets `isCancelled` on its Threads, which then run the functions still + * in the queue before they exit. A long-running function MAY check `isCancelled` to return early. + * A function MUST NOT cancel the Thread that it is given. + * + * A function runs on the calling thread, with a Thread that is never started and never cancelled, + * in two cases: ThreadPool::wait is called while the function is still in the queue, or the size + * of the ThreadPool is `0`. A function MUST NOT `join`, `kill` or `detach` that Thread, and MUST + * NOT rely on `isCancelled` alone to return, because it is never set. + * @remarks A ThreadPool MUST NOT be deallocated by one of its own Threads. + * @extends Object + * @ingroup Concurrency + */ +struct ThreadPool { + + /** + * @brief The superclass. + */ + Object object; + + /** + * @brief The interface type. + * @protected + */ + ThreadPoolInterface *interface[0]; + + /** + * @brief The condition guarding the queue and signaling the state of each ThreadPoolWork. + * @private + */ + Condition *condition; + + /** + * @brief The first ThreadPoolWork in the queue, or `NULL`. + * @private + */ + ThreadPoolWork *head; + + /** + * @brief The number of Threads. + * @remarks Read-only. + */ + size_t size; + + /** + * @brief The last ThreadPoolWork in the queue, or `NULL`. + * @private + */ + ThreadPoolWork *tail; + + /** + * @brief The Threads, `size` in all. + * @private + */ + Array *threads; +}; + +/** + * @brief The ThreadPool interface. + */ +struct ThreadPoolInterface { + + /** + * @brief The superclass interface. + */ + ObjectInterface objectInterface; + + /** + * @fn void ThreadPool::detach(ThreadPool *self, ThreadFunction function, ident data) + * @brief Queues `function` without a handle to wait on. + * @param self The ThreadPool. + * @param function The ThreadFunction. + * @param data The user data, available to `function` as `thread->data`. + * @remarks The return value of `function` is discarded. + * @memberof ThreadPool + */ + void (*detach)(ThreadPool *self, ThreadFunction function, ident data); + + /** + * @fn ThreadPoolWork *ThreadPool::execute(ThreadPool *self, ThreadFunction function, ident data) + * @brief Queues `function`, returning a handle to wait on. + * @param self The ThreadPool. + * @param function The ThreadFunction. + * @param data The user data, available to `function` as `thread->data`. + * @return The ThreadPoolWork, which MUST be passed to ThreadPool::wait. + * @memberof ThreadPool + */ + ThreadPoolWork *(*execute)(ThreadPool *self, ThreadFunction function, ident data); + + /** + * @fn ThreadPool *ThreadPool::init(ThreadPool *self) + * @brief Initializes this ThreadPool with one Thread per online processor. + * @param self The ThreadPool. + * @return The initialized ThreadPool, or `NULL` on error. + * @memberof ThreadPool + */ + ThreadPool *(*init)(ThreadPool *self); + + /** + * @fn ThreadPool *ThreadPool::initWithSize(ThreadPool *self, size_t size) + * @brief Initializes this ThreadPool with the specified number of Threads. + * @param self The ThreadPool. + * @param size The number of Threads. With `0`, every function runs on the calling thread. + * @return The initialized ThreadPool, or `NULL` on error. + * @memberof ThreadPool + */ + ThreadPool *(*initWithSize)(ThreadPool *self, size_t size); + + /** + * @fn ident ThreadPool::wait(ThreadPool *self, ThreadPoolWork *work) + * @brief Waits for the specified ThreadPoolWork to finish, and frees it. + * @param self The ThreadPool. + * @param work The ThreadPoolWork returned by ThreadPool::execute, or `NULL`. + * @return The return value of the ThreadFunction, or `NULL` if `work` is `NULL`. + * @remarks If `work` is still in the queue, it runs on the calling thread. So a ThreadFunction + * MAY wait on other work of its own ThreadPool without a deadlock. + * @memberof ThreadPool + */ + ident (*wait)(ThreadPool *self, ThreadPoolWork *work); +}; + +/** + * @fn Class *ThreadPool::_ThreadPool(void) + * @brief The ThreadPool archetype. + * @return The ThreadPool Class. + * @memberof ThreadPool + */ +OBJECTIVELY_EXPORT Class *_ThreadPool(void); diff --git a/Sources/Objectively/URLSession.c b/Sources/Objectively/URLSession.c index f24b1d2..f4e4c17 100644 --- a/Sources/Objectively/URLSession.c +++ b/Sources/Objectively/URLSession.c @@ -40,11 +40,11 @@ static void dealloc(Object *self) { $(this, invalidateAndCancel); - $(this->locals.thread, join, NULL); + $(this->thread, join, NULL); - release(this->locals.condition); - release(this->locals.thread); - release(this->locals.tasks); + release(this->condition); + release(this->thread); + release(this->tasks); release(this->configuration); super(Object, self, dealloc); @@ -60,10 +60,19 @@ static ident taskWithRequest(URLSession *self, ident task, URLRequest *request, task = $((URLSessionTask *) task, initWithRequestInSession, request, self, completion); if (task) { - synchronized(self->locals.condition, { - $(self->locals.tasks, addObject, task); - $(self->locals.condition, signal); + bool isInvalidated; + + synchronized(self->condition, { + isInvalidated = self->thread->isCancelled; + if (isInvalidated == false) { + $(self->tasks, addObject, task); + $(self->condition, signal); + } }); + + if (isInvalidated) { + task = release(task); + } } return task; @@ -79,6 +88,16 @@ static URLSessionDataTask *dataTaskWithRequest(URLSession *self, URLRequest *req initWithRequestInSession, request, self, completion); if (task) { + bool isInvalidated; + + synchronized(self->condition, { + isInvalidated = self->thread->isCancelled; + }); + + if (isInvalidated) { + return release(task); + } + URLCache *cache = self->configuration->urlCache; if (cache) { task->cachedResponse = $(cache, cachedResponseForRequest, request); @@ -87,10 +106,17 @@ static URLSessionDataTask *dataTaskWithRequest(URLSession *self, URLRequest *req } } - synchronized(self->locals.condition, { - $(self->locals.tasks, addObject, task); - $(self->locals.condition, signal); + synchronized(self->condition, { + isInvalidated = self->thread->isCancelled; + if (isInvalidated == false) { + $(self->tasks, addObject, task); + $(self->condition, signal); + } }); + + if (isInvalidated) { + task = release(task); + } } return task; @@ -151,6 +177,26 @@ static URLSession *init(URLSession *self) { return self; } +/** + * @brief Checks whether any task needs the run loop. + * @return `true` if any task needs the run loop: one that is not idly suspended, and that + * `URLSessionTask::execute` is not running on another thread. + * @remarks The caller MUST hold the condition. + */ +static bool hasWork(const URLSession *self) { + + const Array *tasks = (Array *) self->tasks; + for (size_t i = 0; i < tasks->count; i++) { + + const URLSessionTask *task = $(tasks, objectAtIndex, i); + if (task->isExecuting == false && task->state != URLSESSIONTASK_SUSPENDED) { + return true; + } + } + + return false; +} + /** * @brief ThreadFunction for a URLSession. */ @@ -158,18 +204,18 @@ static ident run(Thread *thread) { URLSession *self = thread->data; - self->locals.handle = curl_multi_init(); - assert(self->locals.handle); + self->handle = curl_multi_init(); + assert(self->handle); while (true) { int ret; bool idle; - synchronized(self->locals.condition, { - while (!thread->isCancelled && self->locals.tasks->count == 0) { - $(self->locals.condition, wait); + synchronized(self->condition, { + while (!hasWork(self) && !(thread->isCancelled && self->tasks->count == 0)) { + $(self->condition, wait); } - idle = thread->isCancelled && self->locals.tasks->count == 0; + idle = thread->isCancelled && self->tasks->count == 0; }); if (idle) { @@ -181,38 +227,57 @@ static ident run(Thread *thread) { for (size_t i = 0; i < tasks->count; i++) { URLSessionTask *task = $(tasks, objectAtIndex, i); + + bool isExecuting; + synchronized(self->condition, { + isExecuting = task->isExecuting; + }); + + if (isExecuting) { + continue; + } + if (task->state == URLSESSIONTASK_SUSPENDING) { - if (task->locals.handle) { - const CURLcode err = curl_easy_pause(task->locals.handle, CURLPAUSE_ALL); + if (task->handle) { + const CURLcode err = curl_easy_pause(task->handle, CURLPAUSE_ALL); assert(err == CURLE_OK); } - task->state = URLSESSIONTASK_SUSPENDED; + synchronized(self->condition, { + if (task->state == URLSESSIONTASK_SUSPENDING) { + task->state = URLSESSIONTASK_SUSPENDED; + } + }); } else if (task->state == URLSESSIONTASK_RESUMING) { - if (task->locals.handle == NULL) { + if (task->handle == NULL) { $(task, setup); - assert(task->locals.handle); + assert(task->handle); - const CURLMcode merr = curl_multi_add_handle(self->locals.handle, task->locals.handle); + const CURLMcode merr = curl_multi_add_handle(self->handle, task->handle); assert(merr == CURLM_OK); } else { - const CURLcode err = curl_easy_pause(task->locals.handle, CURLPAUSE_CONT); + const CURLcode err = curl_easy_pause(task->handle, CURLPAUSE_CONT); assert(err == CURLE_OK); } - task->state = URLSESSIONTASK_RESUMED; + synchronized(self->condition, { + if (task->state == URLSESSIONTASK_RESUMING) { + task->state = URLSESSIONTASK_RESUMED; + } + }); } else if (task->state == URLSESSIONTASK_CANCELING) { - if (task->locals.handle) { - const CURLMcode merr = curl_multi_remove_handle(self->locals.handle, task->locals.handle); + if (task->handle) { + const CURLMcode merr = curl_multi_remove_handle(self->handle, task->handle); assert(merr == CURLM_OK); } + task->result = CURLE_ABORTED_BY_CALLBACK; task->state = URLSESSIONTASK_CANCELED; if (task->completion) { @@ -221,31 +286,31 @@ static ident run(Thread *thread) { $(task, teardown); - synchronized(self->locals.condition, { - $(self->locals.tasks, removeObject, task); + synchronized(self->condition, { + $(self->tasks, removeObject, task); }); - } else if (task->state == URLSESSIONTASK_COMPLETED) { + } else if (task->state == URLSESSIONTASK_CANCELED || task->state == URLSESSIONTASK_COMPLETED) { - synchronized(self->locals.condition, { - $(self->locals.tasks, removeObject, task); + synchronized(self->condition, { + $(self->tasks, removeObject, task); }); } } - CURLMcode merr = curl_multi_wait(self->locals.handle, NULL, 0, 0, NULL); + CURLMcode merr = curl_multi_wait(self->handle, NULL, 0, 0, NULL); assert(merr == CURLM_OK); - merr = curl_multi_perform(self->locals.handle, &ret); + merr = curl_multi_perform(self->handle, &ret); assert(merr == CURLM_OK); CURLMsg *message; - while ((message = curl_multi_info_read(self->locals.handle, &ret))) { + while ((message = curl_multi_info_read(self->handle, &ret))) { URLSessionTask *task = NULL; for (size_t i = 0; i < tasks->count; i++) { URLSessionTask *t = $(tasks, objectAtIndex, i); - if (t->locals.handle == message->easy_handle) { + if (t->handle == message->easy_handle) { task = t; break; } @@ -255,25 +320,35 @@ static ident run(Thread *thread) { if (message->msg == CURLMSG_DONE) { - merr = curl_multi_remove_handle(self->locals.handle, task->locals.handle); + merr = curl_multi_remove_handle(self->handle, task->handle); assert(merr == CURLM_OK); - task->state = URLSESSIONTASK_COMPLETED; + task->result = message->data.result; + + synchronized(self->condition, { + if (task->state == URLSESSIONTASK_CANCELING) { + task->state = URLSESSIONTASK_CANCELED; + } else { + task->state = URLSESSIONTASK_COMPLETED; + } + }); - curl_easy_getinfo(task->locals.handle, CURLINFO_RESPONSE_CODE, (long *) &task->response->httpStatusCode); + curl_easy_getinfo(task->handle, CURLINFO_RESPONSE_CODE, (long *) &task->response->httpStatusCode); - if (message->data.result == CURLE_OK && $((Object *) task, isKindOfClass, _URLSessionDataTask())) { + const bool success = task->state == URLSESSIONTASK_COMPLETED && task->result == CURLE_OK; + + if (success && $((Object *) task, isKindOfClass, _URLSessionDataTask())) { $((URLSessionDataTask *) task, cacheResponse); } if (task->completion) { - task->completion(task, message->data.result == CURLE_OK); + task->completion(task, success); } $(task, teardown); - synchronized(self->locals.condition, { - $(self->locals.tasks, removeObject, task); + synchronized(self->condition, { + $(self->tasks, removeObject, task); }); } } @@ -281,7 +356,7 @@ static ident run(Thread *thread) { release(tasks); } - curl_multi_cleanup(self->locals.handle); + curl_multi_cleanup(self->handle); return NULL; } @@ -298,11 +373,11 @@ static URLSession *initWithConfiguration(URLSession *self, URLSessionConfigurati if (self) { self->configuration = retain(configuration); - self->locals.condition = $(alloc(Condition), init); - self->locals.tasks = $(alloc(Array), init); - self->locals.thread = $(alloc(Thread), initWithFunction, run, self); + self->condition = $(alloc(Condition), init); + self->tasks = $(alloc(Array), init); + self->thread = $(alloc(Thread), initWithFunction, run, self); - $(self->locals.thread, start); + $(self->thread, start); } return self; @@ -314,7 +389,17 @@ static URLSession *initWithConfiguration(URLSession *self, URLSessionConfigurati */ static void invalidateAndCancel(URLSession *self) { - if (self->locals.thread->isCancelled) { + bool isInvalidated; + + synchronized(self->condition, { + isInvalidated = self->thread->isCancelled; + if (isInvalidated == false) { + $(self->thread, cancel); + $(self->condition, signal); + } + }); + + if (isInvalidated) { return; } @@ -327,9 +412,6 @@ static void invalidateAndCancel(URLSession *self) { } release(tasks); - - $(self->locals.thread, cancel); - synchronized(self->locals.condition, $(self->locals.condition, signal)); } static URLSession *_sharedInstance; @@ -357,8 +439,8 @@ static Array *tasks(const URLSession *self) { Array *array; - synchronized(self->locals.condition, { - array = $$(Array, arrayWithArray, (Array *) self->locals.tasks); + synchronized(self->condition, { + array = $$(Array, arrayWithArray, (Array *) self->tasks); }); return array; diff --git a/Sources/Objectively/URLSession.h b/Sources/Objectively/URLSession.h index b0304a6..9864034 100644 --- a/Sources/Objectively/URLSession.h +++ b/Sources/Objectively/URLSession.h @@ -51,6 +51,8 @@ typedef struct URLSessionInterface URLSessionInterface; /** * @brief A management context for loading resources via URLs. + * @remarks Tasks do not retain their session. Deallocating a URLSession cancels its tasks, and a + * task MUST NOT be used after its session is deallocated. * @extends Object * @ingroup URLSession */ @@ -68,34 +70,33 @@ struct URLSession { URLSessionInterface *interface[0]; /** + * @brief The condition, and the Lock guarding access to `tasks`. * @private */ - struct { - /** - * @brief The condition, and the Lock guarding access to `tasks`. - */ - Condition *condition; + Condition *condition; - /** - * @brief The libcurl handle. - */ - ident handle; + /** + * @brief The session configuration. + */ + URLSessionConfiguration *configuration; - /** - * @brief The URLSessionTasks. - */ - Array *tasks; + /** + * @brief The libcurl handle. + * @private + */ + ident handle; - /** - * @brief The backing Thread. - */ - Thread *thread; - } locals; + /** + * @brief The URLSessionTasks. + * @private + */ + Array *tasks; /** - * @brief The session configuration. + * @brief The backing Thread. + * @private */ - URLSessionConfiguration *configuration; + Thread *thread; }; /** @@ -114,7 +115,7 @@ struct URLSessionInterface { * @param self The URLSession. * @param request The URLRequest to perform. * @param completion The completion handler. - * @return The URLSessionDataTask, or `NULL` on error. + * @return The URLSessionDataTask, or `NULL` if this URLSession is invalidated. * @memberof URLSession */ URLSessionDataTask *(*dataTaskWithRequest)(URLSession *self, URLRequest *request, @@ -126,7 +127,7 @@ struct URLSessionInterface { * @param self The URLSession. * @param url The URL to `GET`. * @param completion The completion handler. - * @return The URLSessionDataTask, or `NULL` on error. + * @return The URLSessionDataTask, or `NULL` if this URLSession is invalidated. * @memberof URLSession */ URLSessionDataTask *(*dataTaskWithURL)(URLSession *self, URL *url, @@ -138,7 +139,7 @@ struct URLSessionInterface { * @param self The URLSession. * @param request The URLRequest to perform. * @param completion The completion handler. - * @return The URLSessionDownloadTask, or `NULL` on error. + * @return The URLSessionDownloadTask, or `NULL` if this URLSession is invalidated. * @memberof URLSession */ URLSessionDownloadTask *(*downloadTaskWithRequest)(URLSession *self, URLRequest *request, @@ -150,7 +151,7 @@ struct URLSessionInterface { * @param self The URLSession. * @param url The URL to `GET`. * @param completion The completion handler. - * @return The URLSessionDownloadTask, or `NULL` on error. + * @return The URLSessionDownloadTask, or `NULL` if this URLSession is invalidated. * @memberof URLSession */ URLSessionDownloadTask *(*downloadTaskWithURL)(URLSession *self, URL *url, @@ -179,6 +180,10 @@ struct URLSessionInterface { * @fn void URLSession::invalidateAndCancel(URLSession *self) * @brief Invalidates this URLSession and cancels all pending tasks. * @param self The URLSession. + * @remarks Tasks that `URLSessionTask::execute` is running on other threads are interrupted too. + * After this call, the task factories return `NULL`. This method returns without waiting for the + * cancelled tasks. Deallocating the URLSession invalidates it, then waits for its Thread, which + * finishes once every task has completed or been cancelled. * @memberof URLSession */ void (*invalidateAndCancel)(URLSession *self); @@ -205,7 +210,7 @@ struct URLSessionInterface { * @param self The URLSession. * @param request The URLRequest to perform. * @param completion The completion handler. - * @return The URLSessionUploadTask, or `NULL` on error. + * @return The URLSessionUploadTask, or `NULL` if this URLSession is invalidated. * @memberof URLSession */ URLSessionUploadTask *(*uploadTaskWithRequest)(URLSession *self, URLRequest *request, diff --git a/Sources/Objectively/URLSessionDataTask.c b/Sources/Objectively/URLSessionDataTask.c index 5a0a81d..d8f758b 100644 --- a/Sources/Objectively/URLSessionDataTask.c +++ b/Sources/Objectively/URLSessionDataTask.c @@ -129,19 +129,18 @@ static void execute(URLSessionTask *self) { return; } - $(self, setup); - - CURLcode code = curl_easy_perform(self->locals.handle); - - curl_easy_getinfo(self->locals.handle, CURLINFO_RESPONSE_CODE, (long *) &self->response->httpStatusCode); + URLCache *cache = self->session->configuration->urlCache; + if (cache) { + retain(cache); + } - this->urlSessionTask.state = URLSESSIONTASK_COMPLETED; + super(URLSessionTask, self, execute); - if (code == CURLE_OK) { - $(this, cacheResponse); + if (cache && self->state == URLSESSIONTASK_COMPLETED && self->result == CURLE_OK) { + $(cache, storeCachedResponseForRequest, self->request, self->response, this->data); } - $(self, teardown); + release(cache); } /** @@ -200,8 +199,8 @@ static void setup(URLSessionTask *self) { super(URLSessionTask, self, setup); - curl_easy_setopt(self->locals.handle, CURLOPT_WRITEFUNCTION, writeFunction); - curl_easy_setopt(self->locals.handle, CURLOPT_WRITEDATA, self); + curl_easy_setopt(self->handle, CURLOPT_WRITEFUNCTION, writeFunction); + curl_easy_setopt(self->handle, CURLOPT_WRITEDATA, self); } #pragma mark - Class lifecycle diff --git a/Sources/Objectively/URLSessionDownloadTask.c b/Sources/Objectively/URLSessionDownloadTask.c index 0e4e153..10b3d08 100644 --- a/Sources/Objectively/URLSessionDownloadTask.c +++ b/Sources/Objectively/URLSessionDownloadTask.c @@ -56,8 +56,8 @@ static void setup(URLSessionTask *self) { assert(this->file); - curl_easy_setopt(self->locals.handle, CURLOPT_WRITEFUNCTION, writeFunction); - curl_easy_setopt(self->locals.handle, CURLOPT_WRITEDATA, self); + curl_easy_setopt(self->handle, CURLOPT_WRITEFUNCTION, writeFunction); + curl_easy_setopt(self->handle, CURLOPT_WRITEDATA, self); } #pragma mark - Class lifecycle diff --git a/Sources/Objectively/URLSessionTask.c b/Sources/Objectively/URLSessionTask.c index 3c1c038..709e868 100644 --- a/Sources/Objectively/URLSessionTask.c +++ b/Sources/Objectively/URLSessionTask.c @@ -55,7 +55,6 @@ static void dealloc(Object *self) { } release(this->request); - release(this->session); release(this->response); super(Object, self, dealloc); @@ -69,16 +68,19 @@ static void dealloc(Object *self) { */ static void cancel(URLSessionTask *self) { - switch (self->state) { - case URLSESSIONTASK_RESUMING: - case URLSESSIONTASK_RESUMED: - case URLSESSIONTASK_SUSPENDING: - case URLSESSIONTASK_SUSPENDED: - self->state = URLSESSIONTASK_CANCELING; - break; - default: - break; - } + synchronized(self->session->condition, { + switch (self->state) { + case URLSESSIONTASK_RESUMING: + case URLSESSIONTASK_RESUMED: + case URLSESSIONTASK_SUSPENDING: + case URLSESSIONTASK_SUSPENDED: + self->state = URLSESSIONTASK_CANCELING; + $(self->session->condition, signal); + break; + default: + break; + } + }); } /** @@ -87,13 +89,43 @@ static void cancel(URLSessionTask *self) { */ static void execute(URLSessionTask *self) { + Condition *condition = self->session->condition; + + bool isExecuting = false; + + synchronized(condition, { + switch (self->state) { + case URLSESSIONTASK_CANCELING: + self->state = URLSESSIONTASK_CANCELED; + break; + case URLSESSIONTASK_CANCELED: + case URLSESSIONTASK_COMPLETED: + break; + default: + self->isExecuting = isExecuting = true; + break; + } + }); + + if (isExecuting == false) { + return; + } + $(self, setup); - curl_easy_perform(self->locals.handle); + self->result = curl_easy_perform(self->handle); - curl_easy_getinfo(self->locals.handle, CURLINFO_RESPONSE_CODE, (long *) &self->response->httpStatusCode); + curl_easy_getinfo(self->handle, CURLINFO_RESPONSE_CODE, (long *) &self->response->httpStatusCode); - self->state = URLSESSIONTASK_COMPLETED; + synchronized(condition, { + self->isExecuting = false; + if (self->state == URLSESSIONTASK_CANCELING) { + self->state = URLSESSIONTASK_CANCELED; + } else { + self->state = URLSESSIONTASK_COMPLETED; + } + $(condition, signal); + }); $(self, teardown); } @@ -115,7 +147,7 @@ static URLSessionTask *initWithRequestInSession(URLSessionTask *self, struct URL assert(self->error); self->request = retain(request); - self->session = retain(session); + self->session = session; self->response = $(alloc(URLResponse), init); @@ -133,14 +165,17 @@ static URLSessionTask *initWithRequestInSession(URLSessionTask *self, struct URL */ static void resume(URLSessionTask *self) { - switch (self->state) { - case URLSESSIONTASK_SUSPENDING: - case URLSESSIONTASK_SUSPENDED: - self->state = URLSESSIONTASK_RESUMING; - break; - default: - break; - } + synchronized(self->session->condition, { + switch (self->state) { + case URLSESSIONTASK_SUSPENDING: + case URLSESSIONTASK_SUSPENDED: + self->state = URLSESSIONTASK_RESUMING; + $(self->session->condition, signal); + break; + default: + break; + } + }); } /** @@ -185,14 +220,14 @@ static size_t responseHeader(char *buffer, size_t size, size_t count, void *data /** * @brief The `CURLOPT_XFERINFOFUNCTION`, which updates internal state and dispatches the task's * progress function. - * @remarks This is also the mechanism for resuming suspended tasks. + * @return Nonzero to abort the transfer, which is how `cancel` interrupts `execute`. */ static int progress(ident self, curl_off_t bytesExpectedToReceive, curl_off_t bytesReceived, curl_off_t bytesExpectedToSend, curl_off_t bytesSent) { URLSessionTask *this = (URLSessionTask *) self; - curl_easy_getinfo(this->locals.handle, CURLINFO_RESPONSE_CODE, (long *) &this->response->httpStatusCode); + curl_easy_getinfo(this->handle, CURLINFO_RESPONSE_CODE, (long *) &this->response->httpStatusCode); this->bytesExpectedToReceive = bytesExpectedToReceive; this->bytesExpectedToSend = bytesExpectedToSend; @@ -201,7 +236,12 @@ static int progress(ident self, curl_off_t bytesExpectedToReceive, curl_off_t by this->progress(this); } - return 0; + bool isCanceling; + synchronized(this->session->condition, { + isCanceling = this->state == URLSESSIONTASK_CANCELING; + }); + + return isCanceling; } /** @@ -210,34 +250,35 @@ static int progress(ident self, curl_off_t bytesExpectedToReceive, curl_off_t by */ static void setup(URLSessionTask *self) { - self->locals.handle = curl_easy_init(); - assert(self->locals.handle); + self->handle = curl_easy_init(); + assert(self->handle); - curl_easy_setopt(self->locals.handle, CURLOPT_ERRORBUFFER, self->error); - curl_easy_setopt(self->locals.handle, CURLOPT_FOLLOWLOCATION, true); + curl_easy_setopt(self->handle, CURLOPT_ERRORBUFFER, self->error); + curl_easy_setopt(self->handle, CURLOPT_FOLLOWLOCATION, true); const URLSessionConfiguration *config = self->session->configuration; if (config->connectTimeout) { - curl_easy_setopt(self->locals.handle, CURLOPT_CONNECTTIMEOUT, config->connectTimeout); + curl_easy_setopt(self->handle, CURLOPT_CONNECTTIMEOUT, config->connectTimeout); } if (config->timeout) { - curl_easy_setopt(self->locals.handle, CURLOPT_TIMEOUT, config->timeout); + curl_easy_setopt(self->handle, CURLOPT_TIMEOUT, config->timeout); } if (!config->longPolling) { - curl_easy_setopt(self->locals.handle, CURLOPT_LOW_SPEED_LIMIT, 1L); - curl_easy_setopt(self->locals.handle, CURLOPT_LOW_SPEED_TIME, 10L); + curl_easy_setopt(self->handle, CURLOPT_LOW_SPEED_LIMIT, 1L); + curl_easy_setopt(self->handle, CURLOPT_LOW_SPEED_TIME, 10L); } - curl_easy_setopt(self->locals.handle, CURLOPT_HEADERFUNCTION, responseHeader); - curl_easy_setopt(self->locals.handle, CURLOPT_HEADERDATA, self); + curl_easy_setopt(self->handle, CURLOPT_HEADERFUNCTION, responseHeader); + curl_easy_setopt(self->handle, CURLOPT_HEADERDATA, self); - curl_easy_setopt(self->locals.handle, CURLOPT_XFERINFOFUNCTION, progress); - curl_easy_setopt(self->locals.handle, CURLOPT_XFERINFODATA, self); + curl_easy_setopt(self->handle, CURLOPT_NOPROGRESS, 0L); + curl_easy_setopt(self->handle, CURLOPT_XFERINFOFUNCTION, progress); + curl_easy_setopt(self->handle, CURLOPT_XFERINFODATA, self); Data *body = self->request->httpBody; if (body) { - curl_easy_setopt(self->locals.handle, CURLOPT_POSTFIELDS, body->bytes); - curl_easy_setopt(self->locals.handle, CURLOPT_POSTFIELDSIZE, body->length); + curl_easy_setopt(self->handle, CURLOPT_POSTFIELDS, body->bytes); + curl_easy_setopt(self->handle, CURLOPT_POSTFIELDSIZE, body->length); } struct curl_slist *httpHeaders = NULL; @@ -253,34 +294,34 @@ static void setup(URLSessionTask *self) { $(headers, enumerateObjectsAndKeys, requestHeaders_enumerator, &httpHeaders); } - curl_easy_setopt(self->locals.handle, CURLOPT_HTTPHEADER, httpHeaders); + curl_easy_setopt(self->handle, CURLOPT_HTTPHEADER, httpHeaders); - self->locals.requestHeaders = httpHeaders; + self->requestHeaders = httpHeaders; switch (self->request->httpMethod) { case HTTP_POST: - curl_easy_setopt(self->locals.handle, CURLOPT_POST, true); + curl_easy_setopt(self->handle, CURLOPT_POST, true); break; case HTTP_PUT: - curl_easy_setopt(self->locals.handle, CURLOPT_PUT, true); + curl_easy_setopt(self->handle, CURLOPT_PUT, true); break; case HTTP_DELETE: - curl_easy_setopt(self->locals.handle, CURLOPT_CUSTOMREQUEST, "DELETE"); + curl_easy_setopt(self->handle, CURLOPT_CUSTOMREQUEST, "DELETE"); break; case HTTP_PATCH: - curl_easy_setopt(self->locals.handle, CURLOPT_CUSTOMREQUEST, "PATCH"); + curl_easy_setopt(self->handle, CURLOPT_CUSTOMREQUEST, "PATCH"); break; case HTTP_HEAD: - curl_easy_setopt(self->locals.handle, CURLOPT_NOBODY, true); + curl_easy_setopt(self->handle, CURLOPT_NOBODY, true); break; case HTTP_OPTIONS: - curl_easy_setopt(self->locals.handle, CURLOPT_CUSTOMREQUEST, "OPTIONS"); + curl_easy_setopt(self->handle, CURLOPT_CUSTOMREQUEST, "OPTIONS"); break; default: break; } - curl_easy_setopt(self->locals.handle, CURLOPT_URL, self->request->url->urlString->chars); + curl_easy_setopt(self->handle, CURLOPT_URL, self->request->url->urlString->chars); } /** @@ -289,14 +330,17 @@ static void setup(URLSessionTask *self) { */ static void suspend(URLSessionTask *self) { - switch (self->state) { - case URLSESSIONTASK_RESUMING: - case URLSESSIONTASK_RESUMED: - self->state = URLSESSIONTASK_SUSPENDING; - break; - default: - break; - } + synchronized(self->session->condition, { + switch (self->state) { + case URLSESSIONTASK_RESUMING: + case URLSESSIONTASK_RESUMED: + self->state = URLSESSIONTASK_SUSPENDING; + $(self->session->condition, signal); + break; + default: + break; + } + }); } /** @@ -305,14 +349,14 @@ static void suspend(URLSessionTask *self) { */ static void teardown(URLSessionTask *self) { - if (self->locals.handle) { - curl_easy_cleanup(self->locals.handle); - self->locals.handle = NULL; + if (self->handle) { + curl_easy_cleanup(self->handle); + self->handle = NULL; } - if (self->locals.requestHeaders) { - curl_slist_free_all(self->locals.requestHeaders); - self->locals.requestHeaders = NULL; + if (self->requestHeaders) { + curl_slist_free_all(self->requestHeaders); + self->requestHeaders = NULL; } } diff --git a/Sources/Objectively/URLSessionTask.h b/Sources/Objectively/URLSessionTask.h index 33091ad..bb79a60 100644 --- a/Sources/Objectively/URLSessionTask.h +++ b/Sources/Objectively/URLSessionTask.h @@ -78,21 +78,6 @@ struct URLSessionTask { */ URLSessionTaskInterface *interface[0]; - /** - * @private - */ - struct { - /** - * @brief The backing libcurl handle. - */ - ident handle; - - /** - * @brief HTTP headers, in libcurl list structure. - */ - ident requestHeaders; - } locals; - /** * @brief The count of bytes this task expects to receive. */ @@ -128,6 +113,19 @@ struct URLSessionTask { */ char *error; + /** + * @brief The backing libcurl handle. + * @private + */ + ident handle; + + /** + * @brief `true` while `execute` runs this task on the calling thread. The URLSession does + * not touch the libcurl handle of such a task. + * @private + */ + bool isExecuting; + /** * @brief The progress function. */ @@ -138,13 +136,29 @@ struct URLSessionTask { */ struct URLRequest *request; + /** + * @brief HTTP headers, in libcurl list structure. + * @private + */ + ident requestHeaders; + /** * @brief The response. */ struct URLResponse *response; /** - * @brief The session. + * @brief The libcurl result code (`CURLcode`) of the transfer, `0` on success. + * @remarks Only meaningful once the task is `URLSESSIONTASK_COMPLETED`. A transfer that ends + * early, for example with fewer bytes than its `Content-Length`, completes with a nonzero result. + */ + int result; + + /** + * @brief The session, which is not retained. + * @remarks The session's Thread releases tasks, so a task that retained its session could + * release the session's last reference on that Thread, which then could not join itself. A task + * MUST NOT be used after its session is deallocated. */ struct URLSession *session; @@ -168,6 +182,9 @@ struct URLSessionTaskInterface { * @fn void URLSessionTask::cancel(URLSessionTask *self) * @brief Cancels this task. * @param self The URLSessionTask. + * @remarks This MAY be called from any thread. It also interrupts a task that `execute` is + * running on another thread: `execute` then returns within about one second, and the task is + * `URLSESSIONTASK_CANCELED`. * @memberof URLSessionTask */ void (*cancel)(URLSessionTask *self); @@ -177,7 +194,8 @@ struct URLSessionTaskInterface { * @brief Executes this task synchronously, on the calling thread. * @param self The URLSessionTask. * @memberof URLSessionTask - * @remarks Do not mix calls to execute with calls to resume. + * @remarks Do not mix calls to execute with calls to resume. A task that is cancelled before + * or while it executes ends `URLSESSIONTASK_CANCELED`, otherwise `URLSESSIONTASK_COMPLETED`. */ void (*execute)(URLSessionTask *self); diff --git a/Sources/Objectively/URLSessionUploadTask.c b/Sources/Objectively/URLSessionUploadTask.c index b36c7d9..a583454 100644 --- a/Sources/Objectively/URLSessionUploadTask.c +++ b/Sources/Objectively/URLSessionUploadTask.c @@ -64,10 +64,10 @@ static void setup(URLSessionTask *self) { err = fseek(this->file, 0, SEEK_SET); assert(err == 0); - curl_easy_setopt(self->locals.handle, CURLOPT_INFILESIZE_LARGE, self->bytesExpectedToSend); + curl_easy_setopt(self->handle, CURLOPT_INFILESIZE_LARGE, self->bytesExpectedToSend); - curl_easy_setopt(self->locals.handle, CURLOPT_READFUNCTION, readFunction); - curl_easy_setopt(self->locals.handle, CURLOPT_READDATA, self); + curl_easy_setopt(self->handle, CURLOPT_READFUNCTION, readFunction); + curl_easy_setopt(self->handle, CURLOPT_READDATA, self); } #pragma mark - Class lifecycle diff --git a/Tests/Objectively/.gitignore b/Tests/Objectively/.gitignore index 6857f7c..87767b2 100644 --- a/Tests/Objectively/.gitignore +++ b/Tests/Objectively/.gitignore @@ -22,6 +22,7 @@ Set String StringReader Thread +ThreadPool URL URLSession Vector diff --git a/Tests/Objectively/Makefile.am b/Tests/Objectively/Makefile.am index 111ff10..809111e 100644 --- a/Tests/Objectively/Makefile.am +++ b/Tests/Objectively/Makefile.am @@ -28,6 +28,7 @@ TESTS = \ String \ StringReader \ Thread \ + ThreadPool \ URLCache \ URL \ URLSession \ diff --git a/Tests/Objectively/ThreadPool.c b/Tests/Objectively/ThreadPool.c new file mode 100644 index 0000000..28d1721 --- /dev/null +++ b/Tests/Objectively/ThreadPool.c @@ -0,0 +1,385 @@ +/* + * Objectively: Ultra-lightweight object oriented framework for GNU C. + * Copyright (C) 2014 Jay Dolan + * + * This software is provided 'as-is', without any express or implied + * warranty. In no event will the authors be held liable for any damages + * arising from the use of this software. + * + * Permission is granted to anyone to use this software for any purpose, + * including commercial applications, and to alter it and redistribute it + * freely, subject to the following restrictions: + * + * 1. The origin of this software must not be misrepresented; you must not + * claim that you wrote the original software. If you use this software + * in a product, an acknowledgment in the product documentation would be + * appreciated but is not required. + * + * 2. Altered source versions must be plainly marked as such, and must not be + * misrepresented as being the original software. + * + * 3. This notice may not be removed or altered from any source distribution. + */ + +#include +#include +#include + +#include +#include + +static Condition *condition; +static int counter; +static bool gate; + +static ident addOne(Thread *thread) { + return (ident) ((intptr_t) thread->data + 1); +} + +START_TEST(executeAndWait) { + + ThreadPool *pool = $(alloc(ThreadPool), initWithSize, 2); + ck_assert(pool != NULL); + ck_assert_int_eq(2, pool->size); + + for (intptr_t i = 0; i < 8; i++) { + ThreadPoolWork *work = $(pool, execute, addOne, (ident) i); + ck_assert(work != NULL); + ck_assert_int_eq(i + 1, (intptr_t) $(pool, wait, work)); + } + + ck_assert($(pool, wait, NULL) == NULL); + + release(pool); + +} END_TEST + +static ident incrementAndSignal(Thread *thread) { + + synchronized(condition, { + counter++; + $(condition, signal); + }); + + return NULL; +} + +START_TEST(detach) { + + condition = $(alloc(Condition), init); + counter = 0; + + ThreadPool *pool = $(alloc(ThreadPool), initWithSize, 2); + + for (int i = 0; i < 16; i++) { + $(pool, detach, incrementAndSignal, NULL); + } + + synchronized(condition, { + while (counter < 16) { + $(condition, wait); + } + }); + + ck_assert_int_eq(16, counter); + + release(pool); + release(condition); + +} END_TEST + +static ident currentPthread(Thread *thread) { + return (ident) pthread_self(); +} + +START_TEST(sizeZero) { + + ThreadPool *pool = $(alloc(ThreadPool), initWithSize, 0); + ck_assert_int_eq(0, pool->size); + + ThreadPoolWork *work = $(pool, execute, currentPthread, NULL); + ck_assert(work != NULL); + ck_assert(pthread_equal((pthread_t) $(pool, wait, work), pthread_self())); + + condition = $(alloc(Condition), init); + counter = 0; + + $(pool, detach, incrementAndSignal, NULL); + ck_assert_int_eq(1, counter); + + release(condition); + release(pool); + +} END_TEST + +static ident waitForGate(Thread *thread) { + + synchronized(condition, { + counter++; + $(condition, broadcast); + while (gate == false) { + $(condition, wait); + } + }); + + return (ident) pthread_self(); +} + +static pthread_t detachedThreads[8]; + +static ident recordThread(Thread *thread) { + + synchronized(condition, { + detachedThreads[counter++] = pthread_self(); + $(condition, broadcast); + }); + + return NULL; +} + +START_TEST(queued) { + + condition = $(alloc(Condition), init); + counter = 0; + gate = false; + + ThreadPool *pool = $(alloc(ThreadPool), initWithSize, 1); + + ThreadPoolWork *blocked = $(pool, execute, waitForGate, NULL); + + synchronized(condition, { + while (counter == 0) { + $(condition, wait); + } + }); + + ThreadPoolWork *pending = $(pool, execute, currentPthread, NULL); + ck_assert(pthread_equal((pthread_t) $(pool, wait, pending), pthread_self())); + + synchronized(condition, { + counter = 0; + }); + + for (int i = 0; i < 8; i++) { + $(pool, detach, recordThread, NULL); + } + + synchronized(condition, { + ck_assert_int_eq(0, counter); + gate = true; + $(condition, broadcast); + }); + + const pthread_t worker = (pthread_t) $(pool, wait, blocked); + ck_assert(pthread_equal(worker, pthread_self()) == 0); + + synchronized(condition, { + while (counter < 8) { + $(condition, wait); + } + }); + + for (int i = 0; i < 8; i++) { + ck_assert(pthread_equal(detachedThreads[i], worker)); + } + + release(pool); + release(condition); + +} END_TEST + +static ident waitForGateOrCancel(Thread *thread) { + + synchronized(condition, { + counter++; + $(condition, broadcast); + }); + + while (__atomic_load_n(&thread->isCancelled, __ATOMIC_ACQUIRE) == false) { + usleep(1000); + } + + return NULL; +} + +START_TEST(drainOnDealloc) { + + condition = $(alloc(Condition), init); + counter = 0; + + ThreadPool *pool = $(alloc(ThreadPool), initWithSize, 1); + + $(pool, detach, waitForGateOrCancel, NULL); + + synchronized(condition, { + while (counter == 0) { + $(condition, wait); + } + }); + + for (int i = 0; i < 8; i++) { + $(pool, detach, incrementAndSignal, NULL); + } + + release(pool); + + ck_assert_int_eq(9, counter); + + release(condition); + +} END_TEST + +static pthread_t outerThread, innerThread; + +static ident recordInner(Thread *thread) { + innerThread = pthread_self(); + return (ident) ((intptr_t) thread->data + 1); +} + +static ident waitOnPending(Thread *thread) { + + ThreadPool *pool = thread->data; + outerThread = pthread_self(); + + synchronized(condition, { + counter++; + $(condition, broadcast); + }); + + ThreadPoolWork *work = $(pool, execute, recordInner, (ident) 41); + return $(pool, wait, work); +} + +START_TEST(nestedWait) { + + condition = $(alloc(Condition), init); + counter = 0; + + ThreadPool *pool = $(alloc(ThreadPool), initWithSize, 1); + + ThreadPoolWork *work = $(pool, execute, waitOnPending, pool); + + synchronized(condition, { + while (counter == 0) { + $(condition, wait); + } + }); + + ck_assert_int_eq(42, (intptr_t) $(pool, wait, work)); + ck_assert(pthread_equal(outerThread, pthread_self()) == 0); + ck_assert(pthread_equal(innerThread, outerThread)); + + release(pool); + release(condition); + +} END_TEST + +static ident untilCancelled(Thread *thread) { + + synchronized(condition, { + counter++; + $(condition, broadcast); + }); + + while (__atomic_load_n(&thread->isCancelled, __ATOMIC_ACQUIRE) == false) { + usleep(1000); + } + + synchronized(condition, { + gate = true; + }); + + return NULL; +} + +START_TEST(cancelOnDealloc) { + + condition = $(alloc(Condition), init); + counter = 0; + gate = false; + + ThreadPool *pool = $(alloc(ThreadPool), initWithSize, 1); + + $(pool, detach, untilCancelled, NULL); + + synchronized(condition, { + while (counter == 0) { + $(condition, wait); + } + }); + + release(pool); + + ck_assert(gate); + + release(condition); + +} END_TEST + +static ident increment(Thread *thread) { + + synchronized(condition, { + counter++; + }); + + return NULL; +} + +START_TEST(concurrent) { + + condition = $(alloc(Condition), init); + counter = 0; + + ThreadPool *pool = $(alloc(ThreadPool), initWithSize, 4); + + ThreadPoolWork *work[64]; + for (int i = 0; i < 64; i++) { + work[i] = $(pool, execute, increment, NULL); + } + + for (int i = 0; i < 64; i++) { + $(pool, wait, work[i]); + } + + ck_assert_int_eq(64, counter); + + release(pool); + release(condition); + +} END_TEST + +START_TEST(init) { + + ThreadPool *pool = $(alloc(ThreadPool), init); + ck_assert(pool != NULL); + ck_assert(pool->size > 0); + + release(pool); + +} END_TEST + +int main(int argc, char **argv) { + + TCase *tcase = tcase_create("ThreadPool"); + tcase_add_test(tcase, cancelOnDealloc); + tcase_add_test(tcase, concurrent); + tcase_add_test(tcase, detach); + tcase_add_test(tcase, drainOnDealloc); + tcase_add_test(tcase, executeAndWait); + tcase_add_test(tcase, init); + tcase_add_test(tcase, nestedWait); + tcase_add_test(tcase, queued); + tcase_add_test(tcase, sizeZero); + + Suite *suite = suite_create("ThreadPool"); + suite_add_tcase(suite, tcase); + + SRunner *runner = srunner_create(suite); + + srunner_run_all(runner, CK_VERBOSE); + int failed = srunner_ntests_failed(runner); + + srunner_free(runner); + + return failed; +} diff --git a/Tests/Objectively/URLSession.c b/Tests/Objectively/URLSession.c index 4221aa1..1eb0a1b 100644 --- a/Tests/Objectively/URLSession.c +++ b/Tests/Objectively/URLSession.c @@ -23,6 +23,12 @@ #include #include +#include +#include + +#include +#include +#include #include "Objectively.h" @@ -202,11 +208,225 @@ START_TEST(cachedDataTaskWithURL) { } END_TEST +/** + * @brief Opens a listening socket on the loopback interface that accepts connections, through + * its backlog, but never responds to them. + * @return The socket, and its URL in `url`. + */ +static int silentServer(char *url, size_t size) { + + const int sock = socket(AF_INET, SOCK_STREAM, 0); + ck_assert_int_ge(sock, 0); + + struct sockaddr_in addr = { + .sin_family = AF_INET, + .sin_addr.s_addr = htonl(INADDR_LOOPBACK), + }; + + ck_assert_int_eq(0, bind(sock, (struct sockaddr *) &addr, sizeof(addr))); + ck_assert_int_eq(0, listen(sock, 8)); + + socklen_t len = sizeof(addr); + ck_assert_int_eq(0, getsockname(sock, (struct sockaddr *) &addr, &len)); + + snprintf(url, size, "http://127.0.0.1:%d/", ntohs(addr.sin_port)); + return sock; +} + +static double now(void) { + + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return ts.tv_sec + ts.tv_nsec / 1e9; +} + +static ident invalidateLater(Thread *thread) { + + usleep(500000); + $((URLSession *) thread->data, invalidateAndCancel); + return NULL; +} + +START_TEST(executeCancel) { + + char address[64]; + const int sock = silentServer(address, sizeof(address)); + + URLSession *session = $(alloc(URLSession), init); + URL *url = $(alloc(URL), initWithCharacters, address); + + URLSessionDataTask *task = $(session, dataTaskWithURL, url, NULL); + ck_assert(task != NULL); + + Thread *thread = $(alloc(Thread), initWithFunction, invalidateLater, session); + $(thread, start); + + const double start = now(); + $((URLSessionTask *) task, execute); + const double elapsed = now() - start; + + ck_assert_int_eq(URLSESSIONTASK_CANCELED, task->urlSessionTask.state); + ck_assert(elapsed < 3.0); + + $(thread, join, NULL); + + release(thread); + release(task); + release(url); + release(session); + + close(sock); + +} END_TEST + +START_TEST(resumeCancel) { + + char address[64]; + const int sock = silentServer(address, sizeof(address)); + + condition = $(alloc(Condition), init); + taskDone = false; + taskSuccess = true; + + URLSession *session = $(alloc(URLSession), init); + URL *url = $(alloc(URL), initWithCharacters, address); + + URLSessionDataTask *task = $(session, dataTaskWithURL, url, completion); + ck_assert(task != NULL); + + $((URLSessionTask *) task, resume); + + usleep(200000); + + $((URLSessionTask *) task, cancel); + + const Time timeout = { .tv_sec = 3 }; + Date *date = $$(Date, dateWithTimeSinceNow, &timeout); + + synchronized(condition, { + while (taskDone == false) { + ck_assert($(condition, waitUntilDate, date)); + } + }); + + ck_assert(taskSuccess == false); + ck_assert_int_eq(URLSESSIONTASK_CANCELED, task->urlSessionTask.state); + + release(date); + release(task); + release(url); + release(session); + release(condition); + + close(sock); + +} END_TEST + +START_TEST(suspendCancel) { + + char address[64]; + const int sock = silentServer(address, sizeof(address)); + + URLSession *session = $(alloc(URLSession), init); + URL *url = $(alloc(URL), initWithCharacters, address); + + URLSessionDataTask *task = $(session, dataTaskWithURL, url, NULL); + ck_assert(task != NULL); + + $((URLSessionTask *) task, resume); + usleep(200000); + $((URLSessionTask *) task, suspend); + + const double start = now(); + release(session); + ck_assert(now() - start < 3.0); + + ck_assert_int_eq(URLSESSIONTASK_CANCELED, task->urlSessionTask.state); + + release(task); + release(url); + + close(sock); + +} END_TEST + +static int restStatus; +static Data *restData; + +static void restCompletion(int status, Data *data, void *userData) { + + restStatus = status; + restData = data; +} + +START_TEST(invalidated) { + + URLSession *session = $(alloc(URLSession), init); + $(session, invalidateAndCancel); + + URL *url = $(alloc(URL), initWithCharacters, "http://127.0.0.1:1/"); + ck_assert($(session, dataTaskWithURL, url, NULL) == NULL); + ck_assert($(session, downloadTaskWithURL, url, NULL) == NULL); + + RESTClient *client = $(alloc(RESTClient), initWithSession, session); + + Data *data = (Data *) 0x1; + ck_assert_int_eq(0, $(client, get, "http://127.0.0.1:1/", NULL, &data)); + ck_assert(data == NULL); + + restStatus = -1; + restData = (Data *) 0x1; + $(client, getAsync, "http://127.0.0.1:1/", NULL, restCompletion, NULL); + ck_assert_int_eq(0, restStatus); + ck_assert(restData == NULL); + + release(client); + release(url); + release(session); + +} END_TEST + +START_TEST(restClientCancel) { + + char address[64]; + const int sock = silentServer(address, sizeof(address)); + + URLSession *session = $(alloc(URLSession), init); + RESTClient *client = $(alloc(RESTClient), initWithSession, session); + + Thread *thread = $(alloc(Thread), initWithFunction, invalidateLater, session); + $(thread, start); + + Data *data = NULL; + const double start = now(); + const int status = $(client, get, address, NULL, &data); + const double elapsed = now() - start; + + ck_assert_int_eq(0, status); + ck_assert(data == NULL); + ck_assert(elapsed < 3.0); + + $(thread, join, NULL); + + release(thread); + release(client); + release(session); + + close(sock); + +} END_TEST + int main(int argc, char **argv) { TCase *tcase = tcase_create("URLSession"); + tcase_set_timeout(tcase, 30); tcase_add_test(tcase, asynchronous); tcase_add_test(tcase, cachedDataTaskWithURL); + tcase_add_test(tcase, executeCancel); + tcase_add_test(tcase, invalidated); + tcase_add_test(tcase, restClientCancel); + tcase_add_test(tcase, resumeCancel); + tcase_add_test(tcase, suspendCancel); tcase_add_test(tcase, synchronous); Suite *suite = suite_create("URLSession"); diff --git a/configure.ac b/configure.ac index 1b432b8..f2de154 100644 --- a/configure.ac +++ b/configure.ac @@ -1,6 +1,6 @@ AC_PREREQ(2.69) -AC_INIT([Objectively], [2.2.0], [jay@jaydolan.com]) +AC_INIT([Objectively], [2.3.0], [jay@jaydolan.com]) AC_CONFIG_HEADERS([Sources/Objectively/Config.h]) AC_CONFIG_MACRO_DIR([m4])