diff --git a/.github/workflows/pony-lint.yml b/.github/workflows/pony-lint.yml new file mode 100644 index 0000000..5e4e20b --- /dev/null +++ b/.github/workflows/pony-lint.yml @@ -0,0 +1,24 @@ +name: pony-lint + +on: + pull_request: + paths: + - '**/*.pony' + +concurrency: + group: pony-lint-${{ github.ref }} + cancel-in-progress: true + +permissions: + packages: read + +jobs: + pony-lint: + name: Lint Pony source + runs-on: ubuntu-latest + container: + image: ghcr.io/ponylang/shared-docker-ci-standard-builder:release + steps: + - uses: actions/checkout@v6.0.2 + - name: Lint + run: make lint diff --git a/Makefile b/Makefile index 321c603..86199ab 100755 --- a/Makefile +++ b/Makefile @@ -5,6 +5,7 @@ GET_DEPENDENCIES_WITH := corral fetch CLEAN_DEPENDENCIES_WITH := corral clean COMPILE_WITH := corral run -- ponyc BUILD_DOCS_WITH := corral run -- pony-doc +LINT_WITH := pony-lint BUILD_DIR ?= build/$(config) SRC_DIR := $(PACKAGE) @@ -47,6 +48,9 @@ $(EXAMPLES_BINARIES): $(BUILD_DIR)/%: $(SOURCE_FILES) $(EXAMPLES_SOURCE_FILES) | $(GET_DEPENDENCIES_WITH) $(PONYC) -o $(BUILD_DIR) $(EXAMPLES_DIR)/$* +lint: + $(LINT_WITH) $(SRC_DIR) $(EXAMPLES_DIR) + clean: $(CLEAN_DEPENDENCIES_WITH) rm -rf $(BUILD_DIR) @@ -66,4 +70,4 @@ all: test $(BUILD_DIR): mkdir -p $(BUILD_DIR) -.PHONY: all build-examples clean TAGS test test-one +.PHONY: all build-examples clean lint TAGS test test-one diff --git a/examples/process-array/process-array.pony b/examples/process-array/process-array.pony index 1bc2d85..1867c9f 100644 --- a/examples/process-array/process-array.pony +++ b/examples/process-array/process-array.pony @@ -6,18 +6,25 @@ use "runtime_info" actor Main new create(env: Env) => let array: Array[U8] iso = recover iso [ 201;202;3;4;5;6;7;8;9;10 ] end - let job = fj.Job[Array[U8] iso, USize]( - WorkerBuilder, - Generator(consume array), - AddingCollector(env.out), - SchedulerInfoAuth(env.root)) + let job = + fj.Job[Array[U8] iso, USize]( + WorkerBuilder, + Generator(consume array), + AddingCollector(env.out), + SchedulerInfoAuth(env.root)) job.start() class WorkerBuilder is fj.WorkerBuilder[Array[U8] iso, USize] + """ + Creates `Adder` worker instances. + """ fun ref apply(): fj.Worker[Array[U8] iso, USize] iso^ => Adder class Generator is fj.Generator[Array[U8] iso] + """ + Splits an array into chunks distributed across workers. + """ var _working_set: Array[U8] iso var _distribution_set: Array[USize] = _distribution_set.create() @@ -28,6 +35,9 @@ class Generator is fj.Generator[Array[U8] iso] _distribution_set = fj.EvenlySplitDataElements(_working_set.size(), workers) fun ref apply(): Array[U8] iso^ ? => + """ + Return the next batch of elements, or error when none remain. + """ if _working_set.size() == 0 then error end @@ -37,6 +47,9 @@ class Generator is fj.Generator[Array[U8] iso] consume batch class AddingCollector is fj.Collector[Array[U8] iso, USize] + """ + Sums worker results and prints the total. + """ var _total: USize = 0 let _out: OutStream @@ -52,6 +65,9 @@ class AddingCollector is fj.Collector[Array[U8] iso, USize] _out.print(_total.string()) class Adder is fj.Worker[Array[U8] iso, USize] + """ + Sums the elements of a `U8` array. + """ var _working_set: Array[U8] = _working_set.create() fun ref receive(work_set: Array[U8] iso) => diff --git a/examples/process-array/process_array.pony b/examples/process-array/process_array.pony new file mode 100644 index 0000000..0d0c3ff --- /dev/null +++ b/examples/process-array/process_array.pony @@ -0,0 +1,6 @@ +""" +Demonstrates distributing array processing across multiple workers. + +Each worker sums its portion of the array, and a collector totals +the partial results. +""" diff --git a/examples/random-number-to-string/random-number-to-string.pony b/examples/random-number-to-string/random-number-to-string.pony index e24957e..c201bdc 100644 --- a/examples/random-number-to-string/random-number-to-string.pony +++ b/examples/random-number-to-string/random-number-to-string.pony @@ -6,18 +6,25 @@ use "runtime_info" actor Main new create(env: Env) => - let job = fj.Job[USize, String]( - WorkerBuilder, - Generator, - StringCollector(env.out), - SchedulerInfoAuth(env.root)) + let job = + fj.Job[USize, String]( + WorkerBuilder, + Generator, + StringCollector(env.out), + SchedulerInfoAuth(env.root)) job.start() class WorkerBuilder is fj.WorkerBuilder[USize, String] + """ + Creates USizeToString workers. + """ fun ref apply(): fj.Worker[USize, String] iso^ => USizeToString class Generator is fj.Generator[USize] + """ + Produces random numbers, ending when one is divisible by 1000. + """ let _rand: Rand = _rand.create() fun ref init(workers: USize) => @@ -31,6 +38,9 @@ class Generator is fj.Generator[USize] x class StringCollector is fj.Collector[USize, String] + """ + Collects string results and prints them when the job finishes. + """ let _strings: Array[String] = _strings.create() let _out: OutStream @@ -48,6 +58,9 @@ class StringCollector is fj.Collector[USize, String] end class USizeToString is fj.Worker[USize, String] + """ + Converts a USize to its string representation. + """ var _usize: USize = 0 fun ref receive(work_set: USize) => @@ -55,4 +68,3 @@ class USizeToString is fj.Worker[USize, String] fun ref process(runner: fj.WorkerRunner[USize, String] ref) => runner.deliver(_usize.string()) - diff --git a/examples/random-number-to-string/random_number_to_string.pony b/examples/random-number-to-string/random_number_to_string.pony new file mode 100644 index 0000000..ce12dbc --- /dev/null +++ b/examples/random-number-to-string/random_number_to_string.pony @@ -0,0 +1,3 @@ +""" +Demonstrates fork_join by converting random numbers to strings in parallel. +""" diff --git a/examples/word-count/word-count.pony b/examples/word-count/word-count.pony index 311d2bf..1aa531e 100644 --- a/examples/word-count/word-count.pony +++ b/examples/word-count/word-count.pony @@ -10,15 +10,18 @@ type WordCounts is Map[String, USize] actor Main new create(env: Env) => try - let caps = recover val FileCaps.>set(FileRead).>set(FileStat) end - let fp = FilePath(FileAuth(env.root), env.args(1)?, caps) + let caps = + recover val FileCaps .> set(FileRead) .> set(FileStat) end + let fp = + FilePath(FileAuth(env.root), env.args(1)?, caps) let file = recover iso OpenFile(fp) as File end - let job = fj.Job[String, WordCounts iso]( - WorkerBuilder, - FileReader(consume file), - WordCountTotaler(env.out), - SchedulerInfoAuth(env.root)) + let job = + fj.Job[String, WordCounts iso]( + WorkerBuilder, + FileReader(consume file), + WordCountTotaler(env.out), + SchedulerInfoAuth(env.root)) job.start() else @@ -27,10 +30,16 @@ actor Main end class WorkerBuilder is fj.WorkerBuilder[String, WordCounts iso] + """ + Creates SplitAndCount workers. + """ fun ref apply(): fj.Worker[String, WordCounts iso] iso^ => SplitAndCount class SplitAndCount is fj.Worker[String, WordCounts iso] + """ + Splits input text into words and counts their frequencies. + """ var _working_set: String = "" fun ref receive(data: String) => @@ -39,22 +48,33 @@ class SplitAndCount is fj.Worker[String, WordCounts iso] fun ref process( runner: fj.WorkerRunner[String, WordCounts iso] ref) => - let punctuation = """ !"#$%&'()*+,-./:;<=>?@[\]^_`{|}~‘“ """ + """ + Split input into words, count each, and deliver the result. + """ + let punctuation = + """ !"#$%&'()*+,-./:;<=>?@[\]^_`{|}~'" """ let words_and_counts = recover iso WordCounts end for line in _working_set.split("\n").values() do let cleaned = - recover val line.lower().>lstrip(punctuation) - .>rstrip(punctuation) end + recover val + line.lower() + .> lstrip(punctuation) + .> rstrip(punctuation) + end for word in cleaned.split(punctuation).values() do - words_and_counts.upsert(word, + words_and_counts.upsert( + word, 1, - {(current, provided) => current + provided}) + {(current, provided) => current + provided }) end end runner.deliver(consume words_and_counts) class FileReader is fj.Generator[String] + """ + Reads lines from a file, one per call. + """ let _lines: FileLines var _workers: USize = 0 @@ -68,25 +88,28 @@ class FileReader is fj.Generator[String] _lines.next()? class WordCountTotaler is fj.Collector[String, WordCounts iso] + """ + Merges per-worker word counts and prints the totals. + """ var _counts: (WordCounts | None) = None let _out: OutStream new iso create(out: OutStream) => _out = out - fun ref collect(runner: fj.CollectorRunner[String, WordCounts iso] ref, + fun ref collect( + runner: fj.CollectorRunner[String, WordCounts iso] ref, result: WordCounts iso) => match \exhaustive\ _counts | None => - // We haven't gotten any counts yet, instead of copying the map, let's - // just keep the first one as our base to build upon _counts = consume result | let counts: WordCounts => for (word, count) in (consume result).pairs() do - counts.upsert(word, - count, - {(current, provided) => current + provided}) + counts.upsert( + word, + count, + {(current, provided) => current + provided }) end end @@ -100,4 +123,3 @@ class WordCountTotaler is fj.Collector[String, WordCounts iso] _out.print(word + ":" + count.string()) end end - diff --git a/examples/word-count/word_count.pony b/examples/word-count/word_count.pony new file mode 100644 index 0000000..96d17af --- /dev/null +++ b/examples/word-count/word_count.pony @@ -0,0 +1,3 @@ +""" +Demonstrates fork_join by counting word frequencies across a file in parallel. +""" diff --git a/fork_join/_coordinator.pony b/fork_join/_coordinator.pony index 78cdde4..5513975 100644 --- a/fork_join/_coordinator.pony +++ b/fork_join/_coordinator.pony @@ -19,11 +19,12 @@ actor _Coordinator[Input: Any #send, Output: Any #send] _generator = consume generator _collector_runner = CollectorRunner[Input, Output](consume collector, this) - _max_workers = if max_workers > 0 then - max_workers - else - Scheduler.schedulers(auth).usize() - end + _max_workers = + if max_workers > 0 then + max_workers + else + Scheduler.schedulers(auth).usize() + end _generator.init(_max_workers) @@ -49,9 +50,11 @@ actor _Coordinator[Input: Any #send, Output: Any #send] var mw = _max_workers while true do // create worker - let w = WorkerRunner[Input, Output](this, - _collector_runner, - _worker_builder()) + let w = + WorkerRunner[Input, Output]( + this, + _collector_runner, + _worker_builder()) _workers.set(w) // request data for the worker diff --git a/fork_join/_fail.pony b/fork_join/_fail.pony index b7e2984..740a628 100644 --- a/fork_join/_fail.pony +++ b/fork_join/_fail.pony @@ -4,14 +4,16 @@ use @pony_os_stderr[Pointer[U8]]() primitive _Fail """ - We hit a place in code that we expected that we should be able to reach. + We hit a place in code that we should never be able to reach. """ fun apply(loc: SourceLoc = __loc) => @fprintf( @pony_os_stderr(), - ("This should never happen: failure in %s at line %s\n" + - "Please open an issue at https://github.com/ponylang/fork_join/issues/new") - .cstring(), + ("This should never happen: failure in %s at line %s\n" + + "Please open an issue at " + + "https://github.com/ponylang/fork_join" + + "/issues/new") + .cstring(), loc.file().cstring(), loc.line().string().cstring()) @exit(1) diff --git a/fork_join/_test_collector_terminate.pony b/fork_join/_test_collector_terminate.pony index da6e634..d27c929 100644 --- a/fork_join/_test_collector_terminate.pony +++ b/fork_join/_test_collector_terminate.pony @@ -9,8 +9,9 @@ class \nodoc\ iso _TestCollectorTerminate is UnitTest because the generator has no end and will keep producing data forever. If terminate didn't work then the job should continue running. - This test assumes that `finish` isn't called incorrectly before a job is done. We are reliant on the end-to-end test to provide give us a some - provability of that axiom. + This test assumes that `finish` isn't called incorrectly before a + job is done. We are reliant on the end-to-end test to provide give + us some provability of that axiom. We don't test the results expected as there's not a deterministic number of results that will be collected despite our always sending the termination @@ -23,11 +24,12 @@ class \nodoc\ iso _TestCollectorTerminate is UnitTest h.long_test(1_000_000_000) h.expect_action("collector.finish()") - let job = Job[U8, U8]( - _CollectorTerminateBuilder, - _CollectorTerminateGenerator, - _CollectorTerminateCollector(h), - SchedulerInfoAuth(h.env.root)) + let job = + Job[U8, U8]( + _CollectorTerminateBuilder, + _CollectorTerminateGenerator, + _CollectorTerminateCollector(h), + SchedulerInfoAuth(h.env.root)) job.start() diff --git a/fork_join/_test_end_to_end.pony b/fork_join/_test_end_to_end.pony index 9ca8313..312735b 100644 --- a/fork_join/_test_end_to_end.pony +++ b/fork_join/_test_end_to_end.pony @@ -18,11 +18,12 @@ class \nodoc\ iso _TestEndToEnd is UnitTest let expected: Array[U8] val = recover val [1;3;5;7;9;57;4;3;2;1;88;18] end let input: Array[U8] iso = recover iso expected.clone() end - let job = Job[Array[U8] iso, Array[U8] val]( - _EndToEndBuilder, - _EndToEndGenerator(consume input), - _EndToEndCollector(h, expected), - SchedulerInfoAuth(h.env.root)) + let job = + Job[Array[U8] iso, Array[U8] val]( + _EndToEndBuilder, + _EndToEndGenerator(consume input), + _EndToEndCollector(h, expected), + SchedulerInfoAuth(h.env.root)) job.start() diff --git a/fork_join/_test_job_terminate.pony b/fork_join/_test_job_terminate.pony index be77cc9..d6ef66f 100644 --- a/fork_join/_test_job_terminate.pony +++ b/fork_join/_test_job_terminate.pony @@ -9,8 +9,9 @@ class \nodoc\ iso _TestJobTerminate is UnitTest generator has no end and will keep producing data forever. If terminate didn't work then the job should continue running. - This test assumes that `finish` isn't called incorrectly before a job is done. We are reliant on the end-to-end test to provide give us a some - provability of that axiom. + This test assumes that `finish` isn't called incorrectly before a + job is done. We are reliant on the end-to-end test to provide give + us a some provability of that axiom. We don't test the results expected as there's not a deterministic number of results that will be collected despite our always sending the termination @@ -23,14 +24,13 @@ class \nodoc\ iso _TestJobTerminate is UnitTest h.long_test(1_000_000_000) h.expect_action("collector.finish()") - let job = Job[U8, U8]( + Job[U8, U8]( _JobTerminateBuilder, _JobTerminateGenerator, _JobTerminateCollector(h), SchedulerInfoAuth(h.env.root)) - - job.start() - job.terminate() + .> start() + .> terminate() class \nodoc\ _JobTerminateBuilder is WorkerBuilder[U8, U8] fun ref apply(): Worker[U8, U8] iso^ => diff --git a/fork_join/collector.pony b/fork_join/collector.pony index 0bb56e5..0a8ca54 100644 --- a/fork_join/collector.pony +++ b/fork_join/collector.pony @@ -1,8 +1,9 @@ interface Collector[Input: Any #send, Output: Any #send] """ A `Collector` is the final step in the processing pipeline. The `Collector` - instance receives incremental results from [`Worker`](/fork_join/fork_join-Worker/) - instances and creates a running tabulation. + instance receives incremental results from + [`Worker`](/fork_join/fork_join-Worker/) instances and creates + a running tabulation. When a `fork_join` job is finished, a `finish` message will be sent to the collector so it can take whatever steps are required to communicate the final diff --git a/fork_join/collector_runner.pony b/fork_join/collector_runner.pony index 8eafabb..2a70f60 100644 --- a/fork_join/collector_runner.pony +++ b/fork_join/collector_runner.pony @@ -1,8 +1,9 @@ actor CollectorRunner[Input: Any #send, Output: Any #send] """ - `CollectorRunner` is an actor responsible for receiving messages for a - [`Collector`](/fork_join/fork_join-Collector/) and coordinating job lifecycle with the - other `fork_join` library actors. + `CollectorRunner` is an actor responsible for receiving messages + for a [`Collector`](/fork_join/fork_join-Collector/) and + coordinating job lifecycle with the other `fork_join` library + actors. """ var _terminating: Bool = false let _coordinator: _Coordinator[Input, Output] diff --git a/fork_join/helpers.pony b/fork_join/evenly_split_data_elements.pony similarity index 93% rename from fork_join/helpers.pony rename to fork_join/evenly_split_data_elements.pony index 8c9e0b5..20e5461 100644 --- a/fork_join/helpers.pony +++ b/fork_join/evenly_split_data_elements.pony @@ -11,6 +11,9 @@ primitive EvenlySplitDataElements data_elements to give to a worker corresponding to the bucket. """ fun apply(data_elements: USize, split_across: USize): Array[USize] iso^ => + """ + Split `data_elements` across `split_across` buckets as evenly as possible. + """ // If the number data elements were equal to number of workers to split // across then each worker would get 1 element. // We use division to get a basic value to insert. diff --git a/fork_join/fork_join.pony b/fork_join/fork_join.pony index 920983e..bc1b577 100644 --- a/fork_join/fork_join.pony +++ b/fork_join/fork_join.pony @@ -40,7 +40,8 @@ scheduler threads available. ## User supplied `fork_join` components -To create a `fork_join` job, users must provide an implementation for 4 different interfaces supplied by the `fork_join` library. +To create a `fork_join` job, users must provide an implementation +for 4 different interfaces supplied by the `fork_join` library. ### Worker @@ -150,6 +151,9 @@ out of data (or never if the generator can create an infinite supply of data). A job can be ended early by: -- Sending a [`terminate`](/fork_join/fork_join-Job/#terminate) message to the `Job` instance -- Calling `terminate` from the [`collect`](/fork_join/fork_join-Collector/#collect) method of the `Collector` +- Sending a [`terminate`](/fork_join/fork_join-Job/#terminate) + message to the `Job` instance +- Calling `terminate` from the + [`collect`](/fork_join/fork_join-Collector/#collect) method + of the `Collector` """ diff --git a/fork_join/generator.pony b/fork_join/generator.pony index ac9f32a..0a41e14 100644 --- a/fork_join/generator.pony +++ b/fork_join/generator.pony @@ -1,7 +1,8 @@ interface Generator[A: Any #send] """ - A `Generator` creates data on demand which will be sent to various - [`Worker`](/fork_join/fork_join-Worker/) instances where the data will be processed. + A `Generator` creates data on demand which will be sent to + various [`Worker`](/fork_join/fork_join-Worker/) instances where + the data will be processed. """ fun ref init(workers: USize) """ diff --git a/fork_join/job.pony b/fork_join/job.pony index 5a98766..2f7ca8b 100644 --- a/fork_join/job.pony +++ b/fork_join/job.pony @@ -15,11 +15,13 @@ actor Job[Input: Any #send, Output: Any #send] auth: SchedulerInfoAuth, max_workers: USize = 0) => - _coordinator = _Coordinator[Input, Output](consume worker_builder, - consume generator, - consume collector, - auth, - max_workers) + _coordinator = + _Coordinator[Input, Output]( + consume worker_builder, + consume generator, + consume collector, + auth, + max_workers) be start() => """ diff --git a/fork_join/worker_runner.pony b/fork_join/worker_runner.pony index 44f434d..11a18e1 100644 --- a/fork_join/worker_runner.pony +++ b/fork_join/worker_runner.pony @@ -1,8 +1,9 @@ actor WorkerRunner[Input: Any #send, Output: Any #send] """ `WorkerRunner` is responsible for delivering data to - [Worker](/fork_join/fork_join-Worker/) instances for processing and for coordinating - the job lifecycle with other `fork_join` library actors. + [`Worker`](/fork_join/fork_join-Worker/) instances for processing + and for coordinating the job lifecycle with other `fork_join` + library actors. """ let _coordinator: _Coordinator[Input, Output] let _collector: CollectorRunner[Input, Output]