Concurrency & Parallel Processing

All CyberGo JSON operations are concurrency-safe, with ready-to-use parallel APIs (ParallelIterator, parallel JSONL streaming). This page documents the thread-safety semantics, built-in parallel APIs, and concurrency usage patterns.

Scope split with the Performance page

The "Concurrency" section of Performance shows generic Go patterns (sync.WaitGroup + semaphore + Worker Pool) for manually parallelizing array work; this page documents the library's built-in parallel APIs. They are complementary.

Thread-safety guarantees

Processor is a thread-safe processing engine (source comment: Processor is the main JSON processing engine with thread safety):

• A single Processor instance can be shared across goroutines — all public methods (Get/Set/Delete/Marshal, etc.) are internally protected by atomics and concurrency governance (beginGovernedOp/endGovernedOp).
• Package-level functions (json.Get, json.GetString, etc.) share one global Processor and are concurrency-safe by default.
• A *ParsedJSON from PreParse can be read concurrently — multiple goroutines may call GetFromParsed on the same ParsedJSON simultaneously.

When not to share

The Processor is shareable, but do not share mutable Go containers across goroutines (e.g. handing a map[string]any returned by Get to multiple goroutines for mutation). Returned containers are copies by default (unless CacheSharedResults is on), so mutating a return value does not affect the cache — but concurrent mutation of one container still requires caller-side locking.

ParallelIterator

ParallelIterator parallelizes array processing across CPU cores, with a built-in worker pool, error aggregation, and panic recovery — safer than a hand-rolled goroutine pool.

Basic parallel iteration

package main

import (
	"fmt"
	"sync"

	"github.com/cybergodev/json"
)

func main() {
	data := `{"items":[1,2,3,4,5,6,7,8]}`
	items := json.GetArray(data, "items")

	// Worker count defaults to Config.MaxConcurrency (clamped by array length)
	iter := json.NewParallelIterator(items)
	defer iter.Close()

	var mu sync.Mutex
	var sum int64
	err := iter.ForEach(func(_ int, val any) error {
		mu.Lock()
		sum += int64(val.(float64))
		mu.Unlock()
		return nil
	})
	if err != nil {
		panic(err)
	}
	fmt.Printf("Sum = %d\n", sum)
	// Output: Sum = 36
}

Parallel Map

Map transforms each element in parallel; results preserve input order (each worker writes its own index, no lock needed).

package main

import (
	"fmt"

	"github.com/cybergodev/json"
)

func main() {
	data := `{"items":[1,2,3,4]}`
	items := json.GetArray(data, "items")

	iter := json.NewParallelIterator(items)
	defer iter.Close()

	// Parallel map: each element * 10, result order matches input
	doubled, err := iter.Map(func(_ int, val any) (any, error) {
		return int(val.(float64)) * 10, nil
	})
	if err != nil {
		panic(err)
	}
	fmt.Println(doubled)
	// Output: [10 20 30 40]
}

ParallelIterator API overview

API	Signature	Description
NewParallelIterator	func NewParallelIterator(data []any, cfg ...Config) *ParallelIterator	Creates the iterator; worker count from cfg.MaxConcurrency
ForEach	func (it *ParallelIterator) ForEach(fn func(int, any) error) error	Parallel iteration; returns the first error
ForEachWithContext	func (it *ParallelIterator) ForEachWithContext(ctx context.Context, fn func(int, any) error) error	Supports context cancellation
ForEachBatch	func (it *ParallelIterator) ForEachBatch(batchSize int, fn func(int, []any) error) error	Processes in batches in parallel
Map	func (it *ParallelIterator) Map(transform func(int, any) (any, error)) ([]any, error)	Parallel transform, order-preserving
Filter	func (it *ParallelIterator) Filter(predicate func(int, any) bool) []any	Parallel filter
Close	func (it *ParallelIterator) Close()	Releases resources (call when done)

Full signatures and usage in Iterator types.

Error and panic handling

ForEach stops dispatching new tasks and returns the first error; panics inside workers are recovered and converted to errors, so a callback panic never crashes the process. Use ForEachWithContext for cancellation — it exits gracefully on ctx.Done().

Parallel JSONL streaming

For large JSONL (NDJSON) files, StreamJSONLParallel processes each line across multiple workers.

package main

import (
	"fmt"
	"strings"
	"sync"

	"github.com/cybergodev/json"
)

func main() {
	// Simulated JSONL data (one JSON object per line)
	jsonlData := `{"id":1,"score":95}
{"id":2,"score":82}
{"id":3,"score":78}
{"id":4,"score":90}`

	processor, err := json.New()
	if err != nil {
		panic(err)
	}
	defer processor.Close()

	var mu sync.Mutex
	var total int64
	var count int64

	// 4 workers process lines in parallel
	err = processor.StreamJSONLParallel(strings.NewReader(jsonlData), 4, func(lineNum int, item *json.IterableValue) error {
		score := int64(item.GetInt("score"))
		mu.Lock()
		total += score
		count++
		mu.Unlock()
		return nil
	})
	if err != nil {
		panic(err)
	}
	fmt.Printf("Processed %d records, total score %d\n", count, total)
	// Output: Processed 4 records, total score 345
}

API	Description
StreamJSONLParallel(reader, workers, fn)	Multi-worker parallel JSONL processing
StreamJSONLParallelWithContext(ctx, reader, workers, fn)	Same, with context cancellation/timeout
StreamJSONLChunked(reader, chunkSize, fn)	Chunk-based processing, memory-friendly

Full signatures and config (JSONLWorkers/JSONLChunkSize, etc.) in JSONL processing and JSONL streaming.

Line ordering

In parallel mode the callback lineNum still reflects the original line number, but execution order is not guaranteed. For ordered output, write into a pre-allocated slice at the position matching lineNum.

Global processor for concurrent use

SetGlobalProcessor lets all package-level functions share one custom Processor — suited to multi-goroutine services that need uniform configuration (cache params, hooks, security limits).

package main

import (
	"fmt"
	"sync"

	"github.com/cybergodev/json"
)

func main() {
	// Custom global processor (shared by all package-level functions, concurrency-safe)
	cfg := json.DefaultConfig()
	processor, err := json.New(cfg)
	if err != nil {
		panic(err)
	}
	json.SetGlobalProcessor(processor) // the previous global Processor is closed automatically
	defer json.ShutdownGlobalProcessor() // clean shutdown at app exit

	data := `{"user":{"name":"Alice","age":30}}`

	// Multiple goroutines use package-level functions concurrently (same global Processor)
	var wg sync.WaitGroup
	results := make([]string, 3)
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			switch idx {
			case 0:
				results[idx] = json.GetString(data, "user.name")
			case 1:
				results[idx] = fmt.Sprintf("%d", json.GetInt(data, "user.age"))
			case 2:
				results[idx] = json.GetString(data, "user.name")
			}
		}(i)
	}
	wg.Wait()
	fmt.Println(results)
	// Output: [Alice 30 Alice]
}

Ownership transfer

After SetGlobalProcessor, that Processor's lifecycle is owned by the global — do not call Close() on it manually, or you conflict with the global shutdown logic. Call ShutdownGlobalProcessor() at exit for a clean close and resource release.

MaxConcurrency limit

Config.MaxConcurrency (default 50) is a soft concurrency cap per Processor: an atomic counting semaphore limits in-flight operations. When the cap is reached, new operations return ErrConcurrencyLimit (retryable).

cfg := json.DefaultConfig()
cfg.MaxConcurrency = 100 // raise the per-Processor concurrency cap

• ErrConcurrencyLimit is a retryable transient error (see Error handling).
• Parallel streaming (StreamJSONLParallel) takes its worker count from the explicit argument, not bound directly to MaxConcurrency, but shares the same governance slot.
• ParallelIterator worker count comes from cfg.MaxConcurrency (default 50), clamped by the array length.

Best practices & pitfalls

1. Reuse the Processor; don't create one per request

A Processor holds cache, a recursive processor, and other state — reusing one instance is what makes the cache hit. Calling json.New() per request forfeits cache gains and adds allocations.

2. Sharing the instance is safe; sharing return containers needs care

The Processor is safe across goroutines; but a map/slice returned by Get, if shared for mutation across goroutines, requires caller-side locking (or treat as read-only under CacheSharedResults).

3. Release resources with Close

In long-running services, explicitly defer processor.Close() and defer iter.Close() to avoid cache goroutine and memory leaks. An instance set via SetGlobalProcessor uses ShutdownGlobalProcessor instead.

4. Only parallelize CPU-bound work

Parallelism has scheduling and sync overhead. Small arrays (< ParallelThreshold, default 10) are faster serially; JSONL with many lines and heavy per-line work benefits clearly.

5. Mind ordering in parallel mode

StreamJSONLParallel does not guarantee processing order. For ordered results, write by lineNum into position, then consume in order.

See Also

• Performance — processor reuse, generic Go concurrency patterns, benchmarks
• Iterator types — full ParallelIterator API
• JSONL processing — parallel JSONL API details
• Caching & Pre-Parsing — cache mechanism and PreParse
• Error handling — ErrConcurrencyLimit and error classification