Processes and Concurrency

Imports and Responsibilities

import klyn.process
import klyn.threading

Execute a Child Process

Pass the executable and every argument as separate list entries. The call is synchronous and returns only after the child exits. Standard output and standard error are drained concurrently, so a child producing a large amount on either stream cannot deadlock the other stream.

import klyn.process

result = Process.execute(["klyn", "--version"])

if result.exitCode != 0:
    throw Exception("klyn failed: " + result.error)

print(result.output)
print("child PID: " + result.PID)
print("current PID: " + Process.current.PID)

The optional second argument is written to the child's standard input. Arguments are passed as an argument vector; shell operators, wildcard expansion, and quoting are not interpreted implicitly. Start a shell explicitly only when shell syntax is actually required.

Interactive Programs with PseudoTerminal

Redirected pipes are insufficient for programs that expect a terminal. PseudoTerminal creates a real TTY, supports incremental reads, accepts control sequences, and can be resized in character cells.

import klyn.process

terminal = PseudoTerminal(command="klyn", columns=100, rows=30)
try:
    terminal.write("print(6 * 7)\n")
    Thread.sleep(50)
    print(terminal.readAvailable())
    terminal.resize(120, 36)
    terminal.write("exit\n")
finally:
    terminal.close()

readAvailable() never waits for more data and returns an empty string when nothing is ready. Poll it from a worker or timer in GUI applications; do not block the event thread.

Start and Join Threads

A lambda with no parameters satisfies Runnable. Constructing a thread does not start it; call start(), then join() when the caller requires completion.

import klyn.threading

worker = Thread(
    lambda(): print("background work"),
    "report-worker"
)

worker.start()
worker.join()

assert not worker.isAlive
print(worker.id)
print(worker.state)

join(timeoutMillis) limits how long the caller waits; it does not terminate the worker. Thread.currentThread returns the logical thread currently executing, and Thread.activeCount includes the main thread.

Protect Shared State

Mutex.lock() returns an AutoClosable guard. Prefer the resource form so the lock is released when the block returns or throws.

import klyn.threading

lock = Mutex()
total = 0

try lock.lock():
    total += 10

A semaphore represents permits rather than ownership. It is useful for bounded parallelism and for one thread to signal another.

slots = Semaphore(4)

try slots.acquire():
    # At most four workers may execute this block concurrently.
    print("processing one item")

Atomic Numeric Counters

The atomic type modifier provides atomic numeric updates without a separate lock. It is appropriate for counters, but a mutex is still required when several values form one invariant.

import klyn.threading

completed as atomic = 0
completed++
completed += 9

assert completed == 10
print(completed.value)

Cooperative Cancellation

requestStop() (also named stop()) requests cooperative cancellation. It never kills a native thread. Long-running workers explicitly call Thread.checkpoint() at useful boundaries and return or break when it returns false. Ordinary loops have no compiler-injected polling overhead. Normal resource blocks and finally still perform cleanup.

def processNextBatch() as Void:
    Thread.sleep(10)

def runWorker() as Void:
    while Thread.checkpoint():
        processNextBatch()

worker = Thread(lambda(): runWorker())

worker.start()
# Later, from the controlling thread:
worker.requestStop()
worker.join()

suspend() requests a pause acknowledged at the next checkpoint; it does not suspend arbitrary machine instructions. resume() withdraws that request. A stopped paused worker wakes and its checkpoint returns false. Place checkpoints outside critical sections: pausing while owning a lock can block other workers. isAlive remains true until actual completion, even after stop().

Thread.sleep() cooperates with pause requests and wakes early on stop or interruption. interrupt() is distinct from stopping: it sets the persistent isInterrupted flag without setting stopRequested. After an interrupted sleep, explicitly inspect the flag or the next checkpoint and decide how to exit. Mutex acquisition, semaphore acquisition, join() and arbitrary native I/O are not stop-token-aware. Coordinate their completion explicitly; no request forcibly unlocks a resource or fabricates an acquired semaphore permit. Joining oneself raises IllegalStateException.

Native Thread Priority

priority accepts LOWEST, LOW, NORMAL, HIGH and HIGHEST. A setting made before start() is applied before user code runs; changing a live worker applies it immediately. Windows uses native thread priorities. Linux uses nice adjustments of +10, +5, 0, -5 and -10 relative to the inherited level, bounded by the OS range. Higher priority and even restoring a previously lowered Linux priority may require privileges. An OS refusal raises an exception, not a fictitious successful update. Priority is a scheduling hint and never guarantees which thread executes first.

worker = Thread(lambda(): print("background task"))
worker.priority = "LOW"
worker.start()
worker.join()