### Full Usage Example

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Comprehensive example showing task scheduling, timeout handling, and pool lifecycle management.

```python
from pebble import ProcessPool
from concurrent.futures import TimeoutError

def heavy_computation(n):
    return sum(i**3 for i in range(n))

pool = ProcessPool(max_workers=2, max_tasks=10)

# Schedule with timeout
futures = []
for i in range(20):
    future = pool.schedule(
        heavy_computation,
        args=(i*100000,),
        timeout=5
    )
    futures.append(future)

pool.close()

# Collect results
for i, future in enumerate(futures):
    try:
        result = future.result()
        print(f"Task {i}: {result}")
    except TimeoutError:
        print(f"Task {i}: Timed out")
    except Exception as error:
        print(f"Task {i}: Error {error}")

pool.join()
```

--------------------------------

### Worker Initialization Example

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Demonstrates how to use an initializer function to set up worker processes.

```python
from pebble import ProcessPool

def init_worker(config_path):
    with open(config_path) as f:
        global CONFIG
        CONFIG = json.load(f)

def work_with_config(item):
    return CONFIG['multiplier'] * item

pool = ProcessPool(
    max_workers=4,
    initializer=init_worker,
    initargs=('config.json',)
)

futures = [pool.schedule(work_with_config, args=(i,)) for i in range(10)]
pool.close()
pool.join()
results = [f.result() for f in futures]
```

--------------------------------

### Full Usage Example

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Demonstrates creating a pool, scheduling multiple tasks, and waiting for completion.

```python
from pebble import ThreadPool
import time

def slow_task(x):
    time.sleep(1)
    return x * 2

# Create pool with 4 workers
pool = ThreadPool(max_workers=4)

# Schedule tasks
futures = [pool.schedule(slow_task, args=(i,)) for i in range(10)]

# Close and wait for completion
pool.close()
pool.join()

# Collect results
results = [f.result() for f in futures]
```

--------------------------------

### Context Manager Usage

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Example of using ProcessPool as a context manager.

```python
with ProcessPool(max_workers=4) as pool:
    future = pool.schedule(my_function, timeout=10)
    result = future.result()
```

--------------------------------

### MapFuture Usage Example

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Shows how to map operations, check completion, and cancel remaining work.

```python
from pebble import ThreadPool

pool = ThreadPool(max_workers=4)

# Map returns MapFuture
map_future = pool.map(lambda x: x**2, range(100), chunksize=10)

# Check if map completed
if map_future.done():
    results = map_future.result()
    for result in results:
        print(result)

# Cancel remaining work
if not map_future.done():
    map_future.cancel()

pool.close()
pool.join()
```

--------------------------------

### PebbleFuture Usage Example

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Demonstrates checking if a future should proceed with work using set_running_or_notify_cancel.

```python
from concurrent.futures import Future
from pebble.common import PebbleFuture

future = PebbleFuture()

# Check if we should proceed with work
if future.set_running_or_notify_cancel():
    print("Future is running")
    # Do work
    future.set_result("success")
else:
    print("Future was cancelled before execution")
```

--------------------------------

### Schedule Method Usage Example

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Demonstrates scheduling a task with a timeout and handling potential TimeoutErrors.

```python
from pebble import ProcessPool
from concurrent.futures import TimeoutError

def compute(n):
    return sum(i**2 for i in range(n))

with ProcessPool(max_workers=4) as pool:
    future = pool.schedule(compute, args=(1000000,), timeout=5)
    try:
        result = future.result()
    except TimeoutError:
        print("Computation timed out")
```

--------------------------------

### ThreadPool Map Usage Example

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Demonstrates mapping a function over a range with chunking and handling cancellation.

```python
with ThreadPool() as pool:
    map_future = pool.map(lambda x: x**2, range(100), chunksize=10)
    
    if map_future.cancel():
        print("Map cancelled")
    
    # Iterate over results
    for result in map_future.result():
        print(result)
```

--------------------------------

### Use ProcessFuture for Execution

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/types.md

Comprehensive example showing how to use ProcessFuture for result retrieval, timeouts, and callbacks.

```python
from pebble import concurrent
from concurrent.futures import TimeoutError

@concurrent.process(timeout=5)
def process_data(data_size):
    import time
    time.sleep(2)
    return data_size * 2

# Basic usage
future = process_data(100)

# Wait for completion
result = future.result()  # 200

# Non-blocking check
if future.done():
    if future.exception():
        print(f"Error: {future.exception()}")
    else:
        print(f"Result: {future.result()}")

# With timeout
try:
    result = future.result(timeout=10)
except TimeoutError:
    future.cancel()
    print("Cancelled due to timeout")

# Callback
def on_complete(future):
    try:
        result = future.result()
        print(f"Completed with result: {result}")
    except Exception as error:
        print(f"Failed with error: {error}")

future.add_done_callback(on_complete)
```

--------------------------------

### ProcessPool Map Usage Example

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Demonstrates mapping a function over a range with timeout and chunking, including exception handling.

```python
with ProcessPool(max_workers=4) as pool:
    map_future = pool.map(
        lambda x: x**2,
        range(1000),
        timeout=10,
        chunksize=20
    )
    
    # Iterate and handle results
    for result in map_future.result():
        try:
            print(result)
        except Exception as error:
            print(f"Error: {error}")
```

--------------------------------

### Configuring Multiprocessing Contexts

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Define specific multiprocessing start methods for tasks using the mp_context parameter.

```python
import multiprocessing

# Fork (default on Unix) - fastest, shares memory
ctx_fork = multiprocessing.get_context('fork')

@concurrent.process(mp_context=ctx_fork)
def task():
    return 42

# Spawn (default on Windows) - safest, fresh Python interpreter
ctx_spawn = multiprocessing.get_context('spawn')

@concurrent.process(mp_context=ctx_spawn)
def safe_task():
    return 42

# Forkserver (Unix only) - hybrid approach
ctx_forkserver = multiprocessing.get_context('forkserver')

@concurrent.process(mp_context=ctx_forkserver)
def server_task():
    return 42
```

--------------------------------

### Task pool management with waitforthreads

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-functions.md

A practical example showing how to use waitforthreads to manage a pool of worker threads processing a queue.

```python
from pebble import waitforthreads
import threading
import time
import queue

def process_item(queue_obj, result_queue):
    while True:
        try:
            item = queue_obj.get(timeout=1)
            if item is None:  # Poison pill
                break
            result_queue.put(item * 2)
            queue_obj.task_done()
        except queue.Empty:
            continue

# Create worker threads
work_queue = queue.Queue()
result_queue = queue.Queue()
workers = [
    threading.Thread(target=process_item, args=(work_queue, result_queue))
    for _ in range(4)
]

for worker in workers:
    worker.start()

# Add work
for i in range(100):
    work_queue.put(i)

# Send poison pills
for _ in workers:
    work_queue.put(None)

# Wait for completion
finished = list(waitforthreads(workers, timeout=30))
print(f"All workers finished: {len(finished) == len(workers)}")

# Collect results
results = []
while True:
    try:
        results.append(result_queue.get_nowait())
    except queue.Empty:
        break

print(f"Processed {len(results)} items")
```

--------------------------------

### Unsupported inner scope decoration

Source: https://github.com/noxdafox/pebble/blob/master/doc/index.md

Demonstrates a pattern that is not supported when using spawn or forkserver start methods.

```default
def outer():

    @concurrent.process
    def inner():
        return

    future = inner()

    return future.result()
```

--------------------------------

### Handle ProcessExpired Exception

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/types.md

Example demonstrating how to catch a ProcessExpired exception when a worker process terminates abnormally.

```python
from pebble import ProcessExpired, concurrent
from concurrent.futures import CancelledError, TimeoutError

@concurrent.process
def worker():
    import os
    os._exit(1)  # Abnormal termination

future = worker()

try:
    result = future.result()
except ProcessExpired as error:
    print(f"Process {error.pid} exited with code {error.exitcode}")
    print(f"Reason: {error}")
```

--------------------------------

### View project file structure

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/INDEX.md

Overview of the package directory layout and module responsibilities.

```text
pebble/
├── __init__.py                  # Main package exports
├── concurrent/
│   ├── __init__.py
│   ├── thread.py               # @concurrent.thread
│   └── process.py              # @concurrent.process
├── asynchronous/
│   ├── __init__.py
│   ├── thread.py               # @asynchronous.thread
│   └── process.py              # @asynchronous.process
├── pool/
│   ├── __init__.py
│   ├── base_pool.py            # BasePool, MapFuture, ProcessMapFuture
│   ├── thread.py               # ThreadPool
│   ├── process.py              # ProcessPool
│   └── channel.py              # Worker communication
├── common/
│   ├── __init__.py
│   ├── types.py                # ProcessExpired, ProcessFuture, CONSTS
│   ├── shared.py               # Shared utilities
│   └── process.py              # Process utilities
├── decorators.py               # @synchronized, @sighandler
├── functions.py                # waitforthreads, waitforqueues
└── py.typed                    # Type hints marker
```

--------------------------------

### Configure concurrent.process with multiprocessing contexts

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-concurrent.md

Demonstrates using specific multiprocessing contexts like 'spawn' and integrating with a ProcessPool.

```python
from pebble import concurrent
import multiprocessing

# Use spawn method (safer for some use cases)
ctx = multiprocessing.get_context('spawn')

@concurrent.process(mp_context=ctx)
def isolated_task(x):
    return x * 2

# Use in ProcessPool with specific context
@concurrent.process(pool=pool)
def pooled_task(x):
    return x * 3
```

--------------------------------

### Initialize and Use ThreadPool

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-module-exports.md

Import pool classes and schedule functions for execution within a thread pool.

```python
from pebble.pool import ThreadPool, ProcessPool
# or
from pebble import ThreadPool, ProcessPool

pool = ThreadPool(max_workers=4)
future = pool.schedule(my_function)
```

--------------------------------

### Submit Method Definition

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Compatibility method signature for asyncio integration.

```python
def submit(
    self,
    function: Callable,
    timeout: Optional[float],
    /,
    *args,
    **kwargs
) -> ProcessFuture
```

--------------------------------

### Use sighandler with state management

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-decorators.md

Demonstrates managing application state within signal handlers.

```python
from pebble import sighandler
import signal
import sys

shutdown_event = None

@sighandler(signal.SIGUSR1)
def reload_config(signum, frame):
    global shutdown_event
    print("Reloading configuration...")
    # Reload config logic

@sighandler([signal.SIGINT, signal.SIGTERM])
def graceful_shutdown(signum, frame):
    print("Shutting down gracefully...")
    global shutdown_event
    if shutdown_event:
        shutdown_event.set()

def main_loop(shutdown_event):
    while not shutdown_event.is_set():
        # Main work
        pass

if __name__ == '__main__':
    shutdown_event = threading.Event()
    main_loop(shutdown_event)
```

--------------------------------

### Context Manager Usage

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Demonstrates using the pool as a context manager to ensure resources are cleaned up.

```python
with ThreadPool(max_workers=4) as pool:
    future = pool.schedule(my_function)
    result = future.result()
# Automatically calls close() and join()
```

--------------------------------

### Handle Initializer Failures

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Demonstrates how exceptions raised during worker initialization propagate to the main process when a task is scheduled.

```python
from pebble import ProcessPool

def init_worker(required_file):
    if not os.path.exists(required_file):
        raise RuntimeError(f"Missing {required_file}")
    # Continue initialization

pool = ProcessPool(
    initializer=init_worker,
    initargs=('required.dat',)
)

try:
    future = pool.schedule(task)
    result = future.result()
except RuntimeError as error:
    print(f"Worker initialization failed: {error}")
```

--------------------------------

### Asynchronous process execution

Source: https://github.com/noxdafox/pebble/blob/master/doc/index.md

Demonstrates using the asynchronous.process decorator for asyncio-compatible process execution.

```default
import asyncio

from pebble import asynchronous

@asynchronous.process
def function(arg, kwarg=0):
    return arg + kwarg

async def asynchronous_function():
    result = await function(1, kwarg=1)
    print(result)

asyncio.run(asynchronous_function())
```

--------------------------------

### Initialize Process Pool Workers

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Sets up global state per process by loading configuration files or shared data during worker startup.

```python
from pebble import ProcessPool

# Global per process
_worker_state = None

def init_worker(config_path, shared_data):
    global _worker_state
    # Initialize once per process
    with open(config_path) as f:
        _worker_state = {
            'config': json.load(f),
            'data': shared_data
        }
    print(f"Worker PID {os.getpid()} initialized")

def task(item):
    return process_with_state(item, _worker_state)

pool = ProcessPool(
    max_workers=4,
    initializer=init_worker,
    initargs=('config.json', shared_data)
)
```

--------------------------------

### Executing Map with Timeout

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Demonstrates mapping a function over a range with specific timeout and chunksize configurations.

```python
from pebble import ProcessPool
from concurrent.futures import TimeoutError

def expensive_operation(item):
    import time
    time.sleep(2)
    return item ** 2

pool = ProcessPool(max_workers=4)

# Map with timeout per chunk
map_future = pool.map(
    expensive_operation,
    range(100),
    timeout=10,
    chunksize=20
)

# Iterate over results
try:
    for result in map_future.result():
        print(result)
except TimeoutError:
    print("Chunk processing timed out")
except Exception as error:
    print(f"Processing error: {error}")

pool.close()
pool.join()
```

--------------------------------

### Schedule Method Usage

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Demonstrates scheduling a function and retrieving the result using a future.

```python
from pebble import ThreadPool

def work(x, y):
    return x + y

with ThreadPool(max_workers=4) as pool:
    future = pool.schedule(work, args=(1, 2))
    result = future.result()  # 3
```

--------------------------------

### Instantiate TimeoutError

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/errors.md

Demonstrates the constructor signature for TimeoutError, accepting a message and the timeout duration in seconds.

```python
from concurrent.futures import TimeoutError

# TimeoutError(message, timeout_seconds)
error = TimeoutError('Task Timeout', 5)
# args[0] = 'Task Timeout'
# args[1] = 5  (the timeout duration)
```

--------------------------------

### Importing Pebble Modules and Utilities

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/overview.md

Standard import paths for decorators, pools, utilities, types, and exceptions used within the Pebble library.

```python
# Decorators
from pebble import concurrent, asynchronous
from pebble.concurrent import thread, process
from pebble.asynchronous import thread, process

# Pools
from pebble import ThreadPool, ProcessPool

# Utilities
from pebble import waitforthreads, waitforqueues
from pebble import synchronized, sighandler

# Types
from pebble import ProcessExpired, ProcessFuture, CONSTS

# Exceptions
from concurrent.futures import TimeoutError, CancelledError
from concurrent.futures.process import BrokenProcessPool
```

--------------------------------

### Catching BrokenProcessPool in a ProcessPool

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/errors.md

Demonstrates how to wrap pool scheduling in a try-except block to handle potential pool failures during task submission.

```python
from pebble import ProcessPool
from concurrent.futures import BrokenProcessPool

with ProcessPool(max_workers=2) as pool:
    futures = []
    try:
        for i in range(10):
            future = pool.schedule(lambda x: x)
            futures.append(future)
    except BrokenProcessPool as error:
        print(f"Pool is broken: {error}")
        # All pending tasks will fail with BrokenProcessPool
```

--------------------------------

### Handle Timeout in ProcessPool

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/errors.md

Demonstrates scheduling tasks with a timeout in a ProcessPool and attempting cancellation upon failure.

```python
from pebble import ProcessPool
from concurrent.futures import TimeoutError

def compute(n):
    import time
    time.sleep(n)
    return n ** 2

with ProcessPool() as pool:
    # Schedule with timeout
    future = pool.schedule(compute, args=(10,), timeout=3)
    
    try:
        result = future.result()
    except TimeoutError as error:
        print(f"Computation exceeded {error.args[1]} second timeout")
        
        # Can attempt to cancel
        if future.cancel():
            print("Task cancelled")
```

--------------------------------

### Accessing Internal Configuration Constants

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Import and view the default configuration constants used by the library.

```python
from pebble import CONSTS

CONSTS.sleep_unit         # float = 0.1
CONSTS.term_timeout       # float = 3
CONSTS.channel_lock_timeout # float = 60
```

--------------------------------

### Define Configuration Constants

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/types.md

Defines the internal configuration settings using a dataclass.

```python
@dataclass
class Consts:
    sleep_unit: float = 0.1
    term_timeout: float = 3
    channel_lock_timeout: float = 60

CONSTS = Consts()
```

--------------------------------

### Import asyncio dependencies

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-module-exports.md

Import asyncio components often utilized when integrating Pebble with asynchronous workflows.

```python
import asyncio
asyncio.Future
asyncio.CancelledError
```

--------------------------------

### Map Method Usage

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Demonstrates applying a function to a range of numbers using the map method.

```python
with ThreadPool(max_workers=4) as pool:
    results = pool.map(lambda x: x**2, range(10), chunksize=2)
    for result in results:
        print(result)
```

--------------------------------

### Integrate ProcessPool with AsyncIO

Source: https://github.com/noxdafox/pebble/blob/master/doc/index.md

Demonstrates running blocking functions in a ProcessPool within an asyncio event loop using run_in_executor.

```python
import time
import asyncio

from pebble import ProcessPool

SLEEP = 10
TIMEOUT = 3

def function(seconds):
    print(f"Going to sleep {seconds}s..")
    time.sleep(seconds)
    print(f"Slept {seconds}s.")

async def main():
    loop = asyncio.get_running_loop()
    pool = ProcessPool()

    await loop.run_in_executor(pool, function, TIMEOUT, SLEEP)

asyncio.run(main())
```

--------------------------------

### Map tasks with timeout handling in ProcessPool

Source: https://github.com/noxdafox/pebble/blob/master/doc/index.md

Demonstrates using ProcessPool.map to execute tasks with a timeout and handle potential exceptions like ProcessExpired.

```python
from concurrent.futures import TimeoutError
from pebble import ProcessPool, ProcessExpired

def function(n):
    return n

with ProcessPool() as pool:
    future = pool.map(function, range(100), timeout=10)

    iterator = future.result()

    while True:
        try:
            result = next(iterator)
        except StopIteration:
            break
        except TimeoutError as error:
            print("function took longer than %d seconds" % error.args[1])
        except ProcessExpired as error:
            print("%s. Exit code: %d" % (error, error.exitcode))
        except Exception as error:
            print("function raised %s" % error)
            print(error.traceback)  # Python's traceback of remote process
```

--------------------------------

### Use async/await with concurrent code

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/INDEX.md

Integrate thread or process execution with async/await syntax using the asynchronous module.

```python
@asynchronous.thread
def blocking_call():
    pass

@asynchronous.process
def cpu_intensive():
    pass

async def main():
    result = await blocking_call()
```

--------------------------------

### Use Context Manager for Pool Lifecycle

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Simplifies pool management by automatically handling close and join operations.

```python
from pebble import ProcessPool

with ProcessPool(max_workers=4) as pool:
    futures = [pool.schedule(task) for task in tasks]
    # Automatic close() and join()
```

--------------------------------

### Import concurrent.futures dependencies

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-module-exports.md

Import common concurrency primitives used with Pebble from the concurrent.futures module.

```python
from concurrent.futures import Future, TimeoutError, CancelledError
from concurrent.futures.process import BrokenProcessPool
```

--------------------------------

### Multiplexing multiple queues

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-functions.md

Shows how to use waitforqueues within a worker loop to handle inputs from multiple sources with a specified timeout.

```python
from pebble import waitforqueues
import queue
import threading
import time

def worker(queue_id, target_queue, other_queues):
    while True:
        # Wait for any input
        ready = list(waitforqueues([target_queue] + other_queues, timeout=5))
        
        if not ready:
            print(f"Worker {queue_id}: timeout")
            continue
        
        for q in ready:
            try:
                item = q.get_nowait()
                print(f"Worker {queue_id} processing: {item}")
            except queue.Empty:
                pass

# Create queues for inter-process communication
input_queues = [queue.Queue() for _ in range(3)]
output_queue = queue.Queue()

threads = [
    threading.Thread(target=worker, args=(i, input_queues[i], [output_queue]))
    for i in range(3)
]

for t in threads:
    t.daemon = True
    t.start()

# Send work to different queues
for i, q in enumerate(input_queues):
    q.put(f"task_for_worker_{i}")

time.sleep(1)  # Let threads process
```

--------------------------------

### Compare Decorator and Result Timeouts

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Demonstrates the difference between decorator-level execution limits and result retrieval timeouts.

```python
@concurrent.process(timeout=10)  # Max execution time
def task():
    pass

future = task()

# These are independent:
# - Decorator timeout: kills process after 10 seconds
# - result() timeout: gives up waiting after 2 seconds
try:
    result = future.result(timeout=2)  # Timeout waiting for result
except TimeoutError:
    pass
```

--------------------------------

### Use concurrent.thread for background tasks

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-concurrent.md

Demonstrates decorating functions to run in threads and handling results or timeouts via the returned Future object.

```python
from pebble import concurrent
from concurrent.futures import TimeoutError

@concurrent.thread
def compute(x, y):
    return x + y

@concurrent.thread(name="background_worker", daemon=True)
def long_running_task(duration):
    time.sleep(duration)
    return "done"

# Call the decorated function - returns immediately with a Future
future = compute(5, 3)

# Block until result is available
result = future.result()  # Returns 8

# Handle errors
try:
    result = long_running_task(10).result(timeout=5)
except TimeoutError:
    print("Task took too long")
```

--------------------------------

### Use synchronized with default global lock

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-decorators.md

Demonstrates thread-safe execution using the default module-level lock.

```python
from pebble import synchronized
import threading

# Using default global lock
@synchronized
def update_counter(counter_dict, key):
    counter_dict[key] = counter_dict.get(key, 0) + 1

# Multiple threads can call this, but only one at a time
counter = {}
threads = [
    threading.Thread(target=update_counter, args=(counter, 'count'))
    for _ in range(10)
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print(counter)  # {'count': 10}
```

--------------------------------

### Importing CancelledError

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/errors.md

Shows the standard library imports for CancelledError from concurrent.futures or asyncio.

```python
from concurrent.futures import CancelledError
# or
import asyncio
# asyncio.CancelledError
```

--------------------------------

### Managing Task Cancellation in ProcessPool

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/errors.md

Illustrates how to cancel specific tasks within a ProcessPool and handle the exception during result retrieval.

```python
from pebble import ProcessPool
from concurrent.futures import CancelledError
import time

def long_task(duration):
    time.sleep(duration)
    return f"Slept {duration} seconds"

with ProcessPool() as pool:
    futures = []
    for i in range(10):
        future = pool.schedule(long_task, args=(5,))
        futures.append(future)
    
    # Cancel some tasks
    for i in range(0, 10, 2):
        futures[i].cancel()
    
    time.sleep(1)
    
    for i, future in enumerate(futures):
        try:
            result = future.result()
            print(f"Task {i}: {result}")
        except CancelledError:
            print(f"Task {i}: Cancelled")
```

--------------------------------

### Submit Method Definition

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Defines the alias method for compatibility with asyncio.

```python
def submit(
    self,
    function: Callable,
    *args,
    **kwargs
) -> concurrent.futures.Future
```

--------------------------------

### Monitor multiple queues with waitforqueues

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-functions.md

Demonstrates basic usage of waitforqueues to block until at least one queue in a provided list contains data.

```python
from pebble import waitforqueues
import queue
import threading
import time

def producer(q, item_count):
    for i in range(item_count):
        q.put(f"item_{i}")
        time.sleep(0.5)

# Create queues
queue_a = queue.Queue()
queue_b = queue.Queue()
queue_c = queue.Queue()

# Start producer in background
thread = threading.Thread(target=producer, args=(queue_a, 5))
thread.daemon = True
thread.start()

# Wait for any queue to have items
ready = list(waitforqueues([queue_a, queue_b, queue_c]))
print(f"Ready queues: {len(ready)}")

# Get item from ready queue
if ready:
    item = ready[0].get()
    print(f"Got: {item}")
```

--------------------------------

### Handle system signals

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/overview.md

Uses the @sighandler decorator to register callbacks for specific OS signals.

```python
from pebble import sighandler
import signal
import sys

@sighandler(signal.SIGINT)
def handle_interrupt(signum, frame):
    print("Interrupted")
    sys.exit(0)

@sighandler([signal.SIGTERM, signal.SIGUSR1])
def handle_signals(signum, frame):
    if signum == signal.SIGTERM:
        print("Terminating")
    elif signum == signal.SIGUSR1:
        print("Reloading")

# Now Ctrl+C and kill signals are handled
```

--------------------------------

### Define the __all__ list for the pebble package

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-module-exports.md

The __all__ list explicitly defines the public API of the pebble package.

```python
__all__ = [
    'concurrent',           # submodule
    'asynchronous',         # submodule
    'waitforthreads',       # function
    'waitforqueues',        # function
    'synchronized',         # decorator
    'sighandler',           # decorator
    'ProcessFuture',        # class
    'MapFuture',            # class
    'ProcessMapFuture',     # class
    'ProcessExpired',       # exception
    'ProcessPool',          # class
    'ThreadPool',           # class
    'CONSTS'                # constant object
]
```

--------------------------------

### ProcessFuture

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/types.md

A future subclass for process-based execution with cancellation support.

```APIDOC
## ProcessFuture

### Description
A future subclass for process-based execution with cancellation support.

### Methods
- **cancel() -> bool**: Cancels the process execution if not already finished.
- **result(timeout=None) -> Any**: Blocks until the process completes and returns its result.
- **exception(timeout=None) -> Optional[BaseException]**: Returns the exception raised by the function, or None if successful.
- **running() -> bool**: Returns True if the process is still executing.
- **done() -> bool**: Returns True if the process has finished (success, error, or cancelled).
- **cancelled() -> bool**: Returns True if the process was cancelled.
- **add_done_callback(callable) -> None**: Registers a callback to be called when the process completes.
```

--------------------------------

### Initializing ProcessPool with a Custom Context

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Pass a specific multiprocessing context to the ProcessPool constructor to isolate worker processes.

```python
import multiprocessing
from pebble import ProcessPool

# Use spawn method for isolated workers
ctx = multiprocessing.get_context('spawn')
pool = ProcessPool(max_workers=4, context=ctx)

# All tasks in this pool use spawn method
future = pool.schedule(lambda: 42)
```

--------------------------------

### Catching TypeError in Pebble

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/errors.md

Demonstrates how to catch TypeError exceptions resulting from invalid pool or parameter configurations in decorators.

```python
from pebble import concurrent, ThreadPool, ProcessPool

# Invalid pool type
wrong_pool = "not a pool"

try:
    @concurrent.thread(pool=wrong_pool)
    def task():
        return 42
except TypeError as error:
    print(f"Invalid pool: {error}")

# Invalid parameters
try:
    @concurrent.process(timeout="10")  # Should be float
    def work():
        pass
except TypeError as error:
    print(f"Invalid parameter: {error}")

# Correct usage
pool = ThreadPool(max_workers=4)

@concurrent.thread(pool=pool)
def valid_task():
    return "success"
```

--------------------------------

### Modifying Global Configuration Constants

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Update constants at application startup to affect all subsequently created process pools.

```python
from pebble import CONSTS, ProcessPool

# Access constants
print(CONSTS.sleep_unit)  # 0.1

# Modify for entire application (at startup)
CONSTS.sleep_unit = 0.05  # More responsive but higher CPU
CONSTS.term_timeout = 5   # Longer graceful shutdown period

# Changes affect all new pools
pool = ProcessPool(max_workers=4)  # Uses modified CONSTS
```

--------------------------------

### MapFuture Iteration Behavior

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Demonstrates iterating over results and handling exceptions during map operations.

```python
from pebble import ThreadPool

with ThreadPool(max_workers=4) as pool:
    # map_future.result() returns an iterator
    results = pool.map(lambda x: x**2, range(10))
    
    for result in results:
        # Errors propagate immediately
        try:
            print(result)
        except Exception as error:
            print(f"Error: {error}")
```

--------------------------------

### pebble.waitforqueues(queues, timeout=None)

Source: https://github.com/noxdafox/pebble/blob/master/doc/index.md

Waits for one or more queues to be ready.

```APIDOC
## pebble.waitforqueues(queues, timeout=None)

### Description
Waits for one or more Queue.Queue objects to be ready or until the timeout expires.

### Parameters
- **queues** (list) - Required - A list of Queue.Queue objects.
- **timeout** (float) - Optional - Seconds to block.
```

--------------------------------

### ProcessPool Constructor

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Initializes a new ProcessPool instance to manage worker processes.

```APIDOC
## ProcessPool.__init__

### Description
Initializes a pool of worker processes for scheduling tasks.

### Parameters
- **max_workers** (int) - Optional - Number of worker processes to maintain (default: CPU count)
- **max_tasks** (int) - Optional - Tasks per worker before restart (default: 0, unlimited)
- **initializer** (Optional[Callable]) - Optional - Function called when each worker starts
- **initargs** (list) - Optional - Arguments passed to initializer
- **context** (multiprocessing.context.BaseContext) - Optional - Multiprocessing context (default: multiprocessing)
```

--------------------------------

### ThreadPool Constructor Definition

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Defines the initialization parameters for the ThreadPool class.

```python
class ThreadPool(BasePool):
    def __init__(
        self,
        max_workers: int = multiprocessing.cpu_count(),
        max_tasks: int = 0,
        initializer: Optional[Callable] = None,
        initargs: list = ()
    )
```

--------------------------------

### join

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Waits for all workers to exit.

```APIDOC
## ProcessPool.join

### Description
Waits for all workers to exit.

### Parameters
- **timeout** (Optional[float]) - Optional - Maximum time to wait
```

--------------------------------

### MapFuture result Method

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Signature for the result method.

```python
def result(self, timeout=None) -> MapResults
```

--------------------------------

### Optimize IPC with chunksize

Source: https://github.com/noxdafox/pebble/blob/master/doc/index.md

Uses the chunksize parameter to improve performance for large collections, noting that timeouts apply to the entire chunk.

```python
from pebble import ProcessPool
from multiprocessing import cpu_count
from concurrent.futures import TimeoutError

def function(n):
    return n

elements = list(range(1000000))

cpus = cpu_count()
size = len(elements)
chunksize = size / cpus
# the timeout will be assigned to each chunk
# therefore, we need to consider its size
timeout = 10 * chunksize

with ProcessPool(max_workers=cpus) as pool:
    future = pool.map(function, elements, chunksize=chunksize, timeout=timeout)

    assert list(future.result()) == elements
```

--------------------------------

### Wait for threads or queues

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/INDEX.md

Use waitforthreads or waitforqueues to block until specific resources are ready.

```python
from pebble import waitforthreads, waitforqueues

# Wait for threads to complete
ready_threads = list(waitforthreads(threads, timeout=30))

# Wait for queue items
ready_queues = list(waitforqueues(queues, timeout=5))
```

--------------------------------

### Handle RemoteException in concurrent processes

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/types.md

Demonstrates catching a remote exception and accessing the preserved traceback information.

```python
from pebble import concurrent

@concurrent.process
def failing_function():
    raise ValueError("Invalid value")

future = failing_function()

try:
    future.result()
except ValueError as error:
    # Remote exception preserves traceback
    if hasattr(error, 'traceback'):
        print("Remote process traceback:")
        print(error.traceback)
        
        if hasattr(error, '__cause__'):
            print("Cause:", error.__cause__)
```

--------------------------------

### ThreadPool Constructor

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Initializes a new ThreadPool instance with specified worker and task limits.

```APIDOC
## ThreadPool.__init__

### Description
Initializes a pool of worker threads for scheduling synchronous tasks.

### Parameters
- **max_workers** (int) - Optional - Number of worker threads to maintain (default: CPU count)
- **max_tasks** (int) - Optional - Tasks per worker before restart (default: 0, unlimited)
- **initializer** (Optional[Callable]) - Optional - Function called when each worker starts
- **initargs** (list) - Optional - Arguments passed to initializer
```

--------------------------------

### ProcessPool Constructor Definition

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Defines the initialization parameters for the ProcessPool class.

```python
class ProcessPool(BasePool):
    def __init__(
        self,
        max_workers: int = multiprocessing.cpu_count(),
        max_tasks: int = 0,
        initializer: Optional[Callable] = None,
        initargs: list = (),
        context: multiprocessing.context.BaseContext = multiprocessing
    )
```

--------------------------------

### Define ProcessFuture Methods

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/types.md

Individual method signatures for the ProcessFuture class.

```python
def cancel(self) -> bool
```

```python
def result(self, timeout=None) -> Any
```

```python
def exception(self, timeout=None) -> Optional[BaseException]
```

```python
def running(self) -> bool
```

```python
def done(self) -> bool
```

```python
def cancelled(self) -> bool
```

```python
def add_done_callback(callable) -> None
```

--------------------------------

### Import multiprocessing dependencies

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-module-exports.md

Import core multiprocessing classes required for process management and inter-process communication.

```python
import multiprocessing
multiprocessing.Process
multiprocessing.Pipe
multiprocessing.Queue
multiprocessing.Lock
multiprocessing.context.BaseContext
```

--------------------------------

### Define synchronized function signatures

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-decorators.md

Shows the decorator syntax for using default or custom locks.

```python
@synchronized
def function(*args, **kwargs) -> Any

@synchronized(lock)
def function(*args, **kwargs) -> Any
```

--------------------------------

### Schedule Method Definition

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Defines the signature for scheduling a function to run in the pool.

```python
def schedule(
    self,
    function: Callable,
    args: tuple = (),
    kwargs: dict = {}
) -> concurrent.futures.Future
```

--------------------------------

### ThreadPool Future Usage

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Demonstrates scheduling tasks in a ThreadPool and interacting with the returned Future object.

```python
from pebble import ThreadPool
from concurrent.futures import TimeoutError

pool = ThreadPool(max_workers=4)

# schedule() returns concurrent.futures.Future
future = pool.schedule(lambda x: x**2, args=(5,))

# Standard Future operations
result = future.result()           # 25
exception = future.exception()     # None
is_done = future.done()            # True
is_running = future.running()      # False

pool.close()
pool.join()
```

--------------------------------

### Wait for threads with timeout

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-functions.md

Demonstrates using the timeout parameter to stop waiting for threads after a specified duration.

```python
from pebble import waitforthreads
import threading
import time

def slow_worker(task_id):
    time.sleep(10)

threads = [
    threading.Thread(target=slow_worker, args=(i,))
    for i in range(3)
]

for thread in threads:
    thread.start()

# Wait up to 2 seconds
ready = list(waitforthreads(threads, timeout=2))
print(f"Finished in time: {len(ready)}")
print(f"Still running: {sum(1 for t in threads if t.is_alive())}")
```

--------------------------------

### Iterating and Handling Errors with MapResults

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Demonstrates how to iterate over results from a ThreadPool and handle exceptions raised during processing.

```python
from pebble import ThreadPool

def process(item):
    if item % 2 == 0:
        return item ** 2
    else:
        raise ValueError(f"Odd number: {item}")

pool = ThreadPool(max_workers=4)

results = pool.map(process, range(10), chunksize=2)

# Iterate and handle errors
try:
    for result in results:
        print(f"Result: {result}")
except ValueError as error:
    print(f"Processing error: {error}")

pool.close()
pool.join()
```

--------------------------------

### Handle Execution Results

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/types.md

Shows how to check the status of a result returned by the execute function.

```python
from pebble.common import Result, ResultStatus, execute

def my_function(x):
    if x < 0:
        raise ValueError("Negative value")
    return x ** 2

# execute() returns a Result
result = execute(my_function, 5)
if result.status == ResultStatus.SUCCESS:
    print(f"Result: {result.value}")
elif result.status == ResultStatus.FAILURE:
    print(f"Exception: {result.value}")
else:
    print(f"System error: {result.value}")
```

--------------------------------

### Implement Graceful Pool Shutdown

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Manually manages pool lifecycle by closing and joining with a timeout, falling back to a forced stop if necessary.

```python
from pebble import ProcessPool

pool = ProcessPool(max_workers=4)

# Schedule work
futures = [pool.schedule(expensive_task, args=(i,)) for i in range(100)]

# Close gracefully - allows pending tasks to complete
pool.close()

# Wait for completion with timeout
try:
    pool.join(timeout=30)
    print("All tasks completed")
except TimeoutError:
    print("Some tasks still running after 30 seconds")
    pool.stop()  # Force stop
```

--------------------------------

### Registering Future Callbacks

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Attaching a completion callback to a Future to handle results or exceptions asynchronously.

```python
from pebble import ThreadPool

pool = ThreadPool(max_workers=4)

def on_complete(future):
    try:
        result = future.result()
        print(f"Completed with: {result}")
    except Exception as error:
        print(f"Failed with: {error}")

future = pool.schedule(lambda: 42)
future.add_done_callback(on_complete)

pool.close()
pool.join()
```

--------------------------------

### @concurrent.process

Source: https://github.com/noxdafox/pebble/blob/master/doc/index.md

Runs the decorated function in a concurrent process, returning a ProcessFuture object.

```APIDOC
## @concurrent.process(timeout=None, name=None, daemon=True, context=None, pool=None)

### Description
Runs the decorated function in a concurrent process, taking care of the results and error management. The decorated function returns a pebble.ProcessFuture object.

### Parameters
- **timeout** (float) - Optional - The time in seconds after which the process will be stopped.
- **name** (str) - Optional - The name of the process.
- **daemon** (bool) - Optional - Whether the process should be a daemon process.
- **context** (multiprocessing.context) - Optional - The context object used for starting the process.
- **pool** (pebble.ProcessPool) - Optional - A process pool to use instead of a dedicated process.
```

--------------------------------

### Configure asynchronous.process decorator

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Defines parameters for asynchronous execution within a process.

```python
@asynchronous.process(
    name: Optional[str] = None,
    daemon: bool = True,
    timeout: Optional[float] = None,
    mp_context: Optional[multiprocessing.context.BaseContext] = None,
    pool: Optional[ProcessPool] = None
)
```

--------------------------------

### ProcessMapFuture Class Signature

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-futures.md

Defines the structure of the ProcessMapFuture class and its primary methods.

```python
class ProcessMapFuture(ProcessFuture):
    @property
    def futures(self) -> list
    
    def cancel(self) -> bool
    def result(self, timeout=None) -> MapResults
```

--------------------------------

### @asynchronous.process

Source: https://github.com/noxdafox/pebble/blob/master/doc/index.md

Runs the decorated function in a concurrent process.

```APIDOC
## @asynchronous.process(timeout=None, name=None, daemon=True, context=None, pool=None)

### Description
Runs the decorated function in a concurrent process, returning an asyncio.Future object. Manages results and errors automatically.

### Parameters
- **timeout** (float) - Optional - Time in seconds before the process is stopped and a TimeoutError is raised.
- **name** (str) - Optional - Name of the process.
- **daemon** (bool) - Optional - Whether the process is a daemon process.
- **context** (multiprocessing.context) - Optional - Context object for starting the process.
- **pool** (pebble.ProcessPool) - Optional - Process pool to use; if provided, name, daemon, and context are ignored.
```

--------------------------------

### ThreadPool Lifecycle Methods

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Methods for managing the lifecycle of the thread pool.

```APIDOC
## ThreadPool Lifecycle

### close()
Closes the pool, preventing new task submissions. Pending tasks complete before exit.

### stop()
Stops the pool immediately without completing pending tasks.

### join(timeout=None)
Waits for all workers to exit. Raises TimeoutError if timeout is exceeded.

### active
Property returning True if the pool is running or closed with pending work.
```

--------------------------------

### Define sighandler function signature

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-decorators.md

Shows the decorator syntax for registering a signal handler.

```python
@sighandler(signals)
def function(signum, frame) -> Any
```

--------------------------------

### Run a function with timeout and error handling

Source: https://github.com/noxdafox/pebble/blob/master/README.rst

Executes a function in a separate process with a specified timeout, handling potential TimeoutError or other exceptions.

```python
from pebble import concurrent
from concurrent.futures import TimeoutError

@concurrent.process(timeout=10)
def function(foo, bar=0):
    return foo + bar

future = function(1, bar=2)

try:
    result = future.result()  # blocks until results are ready
except TimeoutError as error:
    print("Function took longer than %d seconds" % error.args[1])
except Exception as error:
    print("Function raised %s" % error)
    print(error.traceback)  # traceback of the function
```

--------------------------------

### Execute blocking functions asynchronously

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-asynchronous.md

Demonstrates using the @asynchronous.thread decorator to run blocking I/O operations without blocking the main event loop.

```python
import asyncio
from pebble import asynchronous

@asynchronous.thread
def blocking_io():
    # Simulate blocking I/O
    time.sleep(2)
    return "result"

@asynchronous.thread(name="io_worker")
def read_file(filename):
    with open(filename, 'r') as f:
        return f.read()

async def main():
    # Await the thread result
    result = await blocking_io()
    print(result)  # "result"
    
    # Multiple concurrent threads
    results = await asyncio.gather(
        blocking_io(),
        blocking_io(),
        blocking_io()
    )
    print(results)

asyncio.run(main())
```

--------------------------------

### Initialize Thread Pool Workers

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Uses threading.local to maintain thread-specific resources like database connections within a ThreadPool.

```python
from pebble import ThreadPool
import threading

# Shared resource per thread
thread_local = threading.local()

def init_thread_worker(db_path):
    # Initialize once per thread
    thread_local.db = connect_database(db_path)
    print(f"Worker initialized: {threading.current_thread().name}")

def task(item):
    # All tasks in this thread use the same connection
    return thread_local.db.query(item)

pool = ThreadPool(
    max_workers=4,
    initializer=init_thread_worker,
    initargs=('database.db',)
)
```

--------------------------------

### Handle system signals

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/INDEX.md

Use the @sighandler decorator to register functions for specific system signals like SIGINT or SIGTERM.

```python
@sighandler([signal.SIGINT, signal.SIGTERM])
def handle_shutdown(signum, frame):
    cleanup()
    sys.exit(0)
```

--------------------------------

### Configure concurrent.process decorator

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Defines parameters for running functions in a separate process.

```python
@concurrent.process(
    name: Optional[str] = None,
    daemon: bool = True,
    timeout: Optional[float] = None,
    mp_context: Optional[multiprocessing.context.BaseContext] = None,
    pool: Optional[ProcessPool] = None
)
```

--------------------------------

### ProcessMapFuture Class Signature

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-pools.md

Defines the structure of the ProcessMapFuture class inheriting from ProcessFuture.

```python
class ProcessMapFuture(ProcessFuture):
    @property
    def futures(self) -> list
```

--------------------------------

### sighandler(signals)

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/api-reference-module-exports.md

A decorator for handling signals.

```APIDOC
## sighandler

### Description
A decorator used to register signal handlers.

### Signature
`(signals) -> decorator`
```

--------------------------------

### ProcessPool.schedule()

Source: https://github.com/noxdafox/pebble/blob/master/_autodocs/configuration.md

Schedules a function for execution in the process pool.

```APIDOC
## pool.schedule(function, args, kwargs, timeout)

### Description
Schedules a single function to be executed by the process pool.

### Parameters
- **function** (Callable) - Required - Function to execute
- **args** (list) - Optional - Positional arguments
- **kwargs** (dict) - Optional - Keyword arguments
- **timeout** (float) - Optional - Max execution time in seconds
```