parl

type AdbAdressProvider ¶ added in v0.4.0

type AdbAdressProvider interface {
	// AdbSocketAddress retrievs the tcp socket address used
	// by a near Adbette implementation
	AdbSocketAddress() (socketAddress AdbSocketAddress)
}

AdressProvider retrieves the address from an adb server or device so that custom devices can be created

type AdbRequest ¶ added in v0.3.0

type AdbRequest string

AdbRequest is a string formatted as an adb request. AdbRequest is only required for implementations using the Adbette protocol impementation

type AdbResponseID ¶ added in v0.4.0

type AdbResponseID string

type AdbSocketAddress ¶ added in v0.4.0

type AdbSocketAddress string

AdbSocketAddress is a tcp socket address accessible to the local host. The format is two parts separated by a colon. The first part is an IP address or hostname. The second part is a numeric port number. The empty string "" represents "localhost:5037". If the port part is missing, such as "localhost" it implies port 5037. If the host part is missing, it implies "localhost". Note that locahost is valid both for IPv4 and IPv6.

type AdbSyncRequest ¶ added in v0.4.0

type AdbSyncRequest string

type Adbette ¶ added in v0.3.0

type Adbette interface {
	// SendReadOkay sends a request to a remote adb endpoint.
	// if anything else than OKAY is received back from the
	// remote endpoint, err is non-nil.
	SendReadOkay(request AdbRequest) (err error)
	// ReadString reads utf-8 text up to 64 KiB-1 in length
	ReadString() (s string, err error)
	// ConnectToDevice sends a forwarding request to an adb
	// server to connect to one of its devices
	ConnectToDevice(serial AndroidSerial) (err error)
	// Shell executes a shell command on a device connected to the adb server.
	//	- out is a combination of stderr and stdout.
	//	- The status code from an on-device command cannot be obtained
	Shell(command string) (out []byte, err error)
	// ShellStream executes a shell command on the device returning a readable socket
	ShellStream(command string) (conn io.ReadWriteCloser, err error)
	// TrackDevices orders a server to emit serial number as they become available
	TrackDevices() (err error)
	// Devices lists the currently online serials
	Devices() (serials []AndroidSerial, err error)
	// DeviceStati returns all available serials and their status
	// The two slices correspond and are of the same length
	DeviceStati() (serials []AndroidSerial, stati []AndroidStatus, err error)
	// Closer closes an adb connection, meant to be a deferred function
	Closer(errp *error)
	// SetSync sets the protocol mode of an adb device connection to sync
	SetSync() (err error)
	// LIST is a sync request that lists file system entries in a directory of an adb device
	LIST(remoteDir string, dentReceiver func(mode uint32, size uint32, time uint32, byts []byte) (err error)) (err error)
	// RECV fetches the contents of a file on an adb device
	RECV(remotePath string, blobReceiver func(data []byte) (err error)) (err error)
	// CancelError is a value that a LIST or RECV callback routines can return to
	// cancel further invocations
	CancelError() (cancelError error)

	SendBlob(syncRequest AdbSyncRequest, data []byte) (err error)
	ReadBlob() (byts []byte, err error)
	ReadResponseID() (responseID AdbResponseID, err error)
	ReadBytes(byts []byte) (err error)
	SendBytes(byts []byte) (err error)
}

Adbette is a minimal implementation of the adb Android debug bridge protocol. Adbette include both adb server and Android device functions. Adbette is extensible in that additional protocol features are easily implemented without concern for protocol details. to shutdown an Adbette and release its resouces, invoke the Closer method

type Adbetter ¶ added in v0.3.0

type Adbetter interface {
	NewConnection(address AdbSocketAddress, ctx context.Context) (conn Adbette, err error)
}

Adbetter is a factory instance for connections featuring Adbette context is used for terminating a connection attempt. context is not retained in the connection.

type AddError ¶ added in v0.4.106

type AddError func(err error)

AddError is a function to submit non-fatal errors

var NoAddErr AddError

type AddErrorIf ¶ added in v0.4.106

type AddErrorIf interface {
	// AddError is a function to submit non-fatal errors
	AddError(err error)
}

AddErrorIf is an interface implementing the AddError function

type AggregatePriority ¶ added in v0.4.30

type AggregatePriority[V any, P constraints.Ordered] interface {
	// Aggregator returns the aggregator associated with this AggregatePriority
	Aggregator() (aggregator Aggregator[V, P])
	// Update caches the current priority from the aggregator
	Update()
	// Priority returns the effective cached priority
	//	- Priority is used by consumers of the AggregatingPriorityQueue
	CachedPriority() (priority P)
	// Index indicates insertion order
	//	- used for ordering elements of equal priority
	Index() (index int)
}

AggregatePriority caches the priority value from an aggregator for priority.

• V is the value type used as a pointer
• P is the priority type descending order, ie. Integer Floating-Point string

type AggregatingPriorityQueue ¶ added in v0.4.30

type AggregatingPriorityQueue[V any, P constraints.Ordered] interface {
	// Get retrieves a possible value container associated with valuep
	Get(valuep *V) (aggregator Aggregator[V, P], ok bool)
	// Put stores a new value container associated with valuep
	//   - the valuep is asusmed to not have a node in the queue
	Put(valuep *V, aggregator Aggregator[V, P])
	// Update re-prioritizes a value
	Update(valuep *V)
	// Clear empties the priority queue. The hashmap is left intact.
	Clear()
	// List returns the first n or default all values by pirority
	List(n ...int) (aggregatorQueue []AggregatePriority[V, P])
}

AggregatingPriorityQueue uses cached priority obtained from Aggregators that operates on the values outside of the AggregatingPriorityQueue.

• the Update method reprioritizes an updated value element

type Aggregator ¶ added in v0.4.30

type Aggregator[V any, P constraints.Ordered] interface {
	// Value returns the value object this Aggregator is associated with
	//	- the Value method is used by consumers of the AggregatingPriorityQueue
	Value() (valuep *V)
	// Aggregate aggregates and snapshots data values from the value object.
	//	- Aggregate is invoked outside of AggregatingPriorityQueue
	Aggregate()
	// Priority returns the current priority for the associated value
	//	- this priority is cached by AggregatePriority
	Priority() (priority P)
}

Aggregator aggregates, snapshots and assigns priority to an associated value.

• V is the value type used as a pointer
• V may be a thread-safe object whose values change in real-time
• P is the priority type descending order, ie. Integer Floating-Point string

type AndroidSerial ¶ added in v0.3.0

type AndroidSerial string

AndroidSerial uniquely identities an Android device.

• has .String and .IsValid, is Ordered
• typically a string of a dozen or so 8-bit chanacters consisting of lower and upper case a-zA-Z0-9

func NewAndroidSerial(s string) (serial AndroidSerial)

func (a AndroidSerial) IsValid() (isValid bool)

func (a AndroidSerial) String() (s string)

type AndroidStatus ¶ added in v0.3.0

type AndroidStatus string

AndroidStatus indicates the current status of a device known to a Server or Serverette

• AndroidStatus is a single word of ANSII-set characters

const AndroidOnline AndroidStatus = "device"

func NewAndroidStatus(s string) (status AndroidStatus)

func (a AndroidStatus) IsOnline() (isOnline bool)

func (a AndroidStatus) IsValid() (isValid bool)

func (a AndroidStatus) String() (s string)

type AssignedPriority ¶ added in v0.4.30

type AssignedPriority[V any, P constraints.Ordered] interface {
	SetPriority(priority P)
}

AssignedPriority contains the assigned priority for a priority-queue element

• V is the element value type whose pointer-value provides identity
• P is the priority, a descending-ordered type: Integer Floating-Point string

type Atomic32 ¶ added in v0.4.143

type Atomic32[T uint32Types] struct {
	// contains filtered or unexported fields
}

Atomic32 is a generic 32-bit integer with atomic access

• generic for named types of select signed and unsigned underlying integers
• generic for select built-in integer types
• includes ~int ~uint
• excludes ~uint64 ~uintptr ~int64
• for large values or excluded types, use Atomic64
• generic version of atomic.Uint32
• when using int or uint underlying type on a 64-bit platform, type-conversion data loss may occur for larger than 32-bit values
• no performance impact compared to other atomics

func (a *Atomic32[T]) Add(delta T) (new T)

func (a *Atomic32[T]) CompareAndSwap(old, new T) (swapped bool)

func (a *Atomic32[T]) Load() (value T)

func (a *Atomic32[T]) Store(val T)

func (a *Atomic32[T]) Swap(new T) (old T)

type Atomic64 ¶ added in v0.4.148

type Atomic64[T constraints.Integer] struct{ atomic.Uint64 }

Atomic64 is a generic 64-bit integer with atomic access

• generic for named types of any underlying integer or any built-in integer type
• generic version of atomic.Uint64
• —
• go1.21.5 due to alignment using atomic.align64, Atomic64 must be based on atomic.Uint64

func (a *Atomic64[T]) Add(delta T) (new T)

func (a *Atomic64[T]) CompareAndSwap(old, new T) (swapped bool)

func (a *Atomic64[T]) Load() (value T)

func (a *Atomic64[T]) Store(val T)

func (a *Atomic64[T]) Swap(new T) (old T)

type AtomicCounter ¶ added in v0.4.41

type AtomicCounter atomic.Uint64

AtomicCounter is a uint64 thread-safe counter

• atomic.Uint64 added:
• Inc Dec delegating to Add
• Inc2 Dec2: preventing wrap-around CompareAndSwap mechanic
• Set sets particular value
• Value returns current value

func (a *AtomicCounter) Add(value uint64) (newValue uint64)

func (a *AtomicCounter) Dec() (value uint64)

func (a *AtomicCounter) Dec2() (value uint64, didDec bool)

func (a *AtomicCounter) Inc() (value uint64)

func (a *AtomicCounter) Inc2() (value uint64, didInc bool)

func (a *AtomicCounter) Set(value uint64) (oldValue uint64)

func (a *AtomicCounter) Value() (value uint64)

type AtomicMax ¶ added in v0.4.41

type AtomicMax[T constraints.Integer] struct {
	// contains filtered or unexported fields
}

AtomicMax is a thread-safe max container

• hasValue indicator true if a value was equal to or greater than threshold
• optional threshold for minimum accepted max value
• generic for any basic or named Integer type
• negative values cause panic
• can used to track maximum time.Duration that should never be negative
• if threshold is not used, initialization-free
• —
• wait-free CompareAndSwap mechanic

func NewAtomicMax[T constraints.Integer](threshold T) (atomicMax *AtomicMax[T])

func (m *AtomicMax[T]) Max() (value T, hasValue bool)

func (m *AtomicMax[T]) Max1() (value T)

func (m *AtomicMax[T]) Value(value T) (isNewMax bool)

type AtomicMin ¶ added in v0.4.43

type AtomicMin[T constraints.Integer] struct {
	// contains filtered or unexported fields
}

AtomicMin is a thread-safe container for a minimum value of any integer type

• hasValue indicator
• generic for any underlying Integer type
• if type is signed, min may be negative
• lock for first Value invocation
• initialization-free

func (a *AtomicMin[T]) Min() (value T, hasValue bool)

func (a *AtomicMin[T]) Value(value T) (isNewMin bool)

type Awaitable ¶ added in v0.4.115

type Awaitable struct {
	// contains filtered or unexported fields
}

Awaitable is a semaphore allowing any number of threads to observe and await an event

• one-to-many, synchronizes-before, initialization-free
• the synchronization mechanic is closing channel, allowing consumers to await multiple events
• Awaitable.IsClosed provides thread-safe observability
• Awaitable.Close is idempotent, thread-safe, deferrable, initialization-free and panic-free
• parl.CyclicAwaitable is re-armable, cyclic version
• —
• alternative low-blocking inter-thread mechanics are sync.WaitGroup and sync.RWMutex but those are less performant for the managing thread

func (a *Awaitable) Ch() (ch AwaitableCh)

func (a *Awaitable) Close() (didClose bool)

func (a *Awaitable) IsClosed() (isClosed bool)

type AwaitableCh ¶ added in v0.4.115

type AwaitableCh <-chan struct{}

AwaitableCh is a one-to-many inter-thread wait-mechanic with happens-before

• AwaitableCh implements a semaphore
• implementation is a channel whose only allowed operation is channel receive
• AwaitableCh transfers no data, instead channel close is the significant event

Usage:

<-ch // waits for event

select {
  case <-ch:
    hasHappened = true
  default:
    hasHappened = false
}

type CBFunc ¶ added in v0.4.41

type CBFunc func(reason CBReason, maxParallelism uint64, maxLatency time.Duration, threadID ThreadID)

CBFunc is a thread-safe function invoked on

• reason == ITParallelism: parallelism exceeding parallelismWarningPoint
• reason == ITLatency: latency of an ongoing or just ended invocation exceeds latencyWarningPoint

type CBReason ¶ added in v0.4.41

type CBReason uint8

CBReason explains to consumer why parl.InvocationTimer invoked the callback

• ITParallelism ITLatency

const (
	// [parl.InvocationTimer] callback due to increased parallelism
	ITParallelism CBReason = iota + 1
	// [parl.InvocationTimer] callback due to increased latency
	ITLatency
)

func (r CBReason) String() (s string)

type CbSlowDetector ¶ added in v0.4.44

type CbSlowDetector func(sdi *SlowDetectorInvocation, hasReturned bool, duration time.Duration)

type Certificate ¶ added in v0.4.26

type Certificate interface {
	DER() (der CertificateDer)
	PEM() (pemBytes PemBytes)
	ParseCertificate() (certificate *x509.Certificate, err error)
}

type CertificateAuthority ¶ added in v0.4.26

type CertificateAuthority interface {
	Check() (cert *x509.Certificate, err error) // gets x509.Certificate version
	DER() (certificateDer CertificateDer)       // untyped bytes, der: Distinguished Encoding Rules binary format
	Sign(template *x509.Certificate, publicKey crypto.PublicKey) (certDER CertificateDer, err error)
	PEM() (pemBytes PemBytes)
	Private() (privateKey PrivateKey)
}

type CertificateDer ¶ added in v0.4.26

type CertificateDer []byte

CertificateDer is a binary encoding of a certificate. der: Distinguished Encoding Rules is a binary format based on asn1.

type ClosableChan ¶ added in v0.4.0

type ClosableChan[T any] struct {
	// contains filtered or unexported fields
}

ClosableChan wraps a channel with thread-safe idempotent panic-free observable close.

• ClosableChan is initialization-free
• Close is deferrable
• IsClosed provides wether the channel is closed

Usage:

var errCh parl.ClosableChan[error]
go thread(&errCh)
err, ok := <-errCh.Ch()
if errCh.IsClosed() { // can be inspected
…

func thread(errCh *parl.ClosableChan[error]) {
  var err error
  …
  defer errCh.Close(&err) // will not terminate the process
  errCh.Ch() <- err

func NewClosableChan[T any](ch ...chan T) (closable *ClosableChan[T])

func (c *ClosableChan[T]) Ch() (ch chan T)

func (cl *ClosableChan[T]) Close(errp ...*error) (didClose bool, err error)

func (c *ClosableChan[T]) IsClosed(includePending ...bool) (isClosed bool)

func (c *ClosableChan[T]) ReceiveCh() (ch <-chan T)

func (c *ClosableChan[T]) SendCh() (ch chan<- T)

type Counter ¶ added in v0.3.0

type Counter interface {
	// Inc increments the counter. Thread-Safe, method chaining
	Inc() (counter Counter)
	// Dec decrements the counter but not below zero. Thread-Safe, method chaining
	Dec() (counter Counter)
	// Add adds a positive or negative delta. Thread-Safe, method chaining
	Add(delta int64) (counter Counter)
}

Counter is the data provider interface for a counter

• Inc Dec Add operations, Thread-safe

type CounterID ¶ added in v0.3.0

type CounterID string

CounterID is a unique type containing counter names

type CounterSet ¶ added in v0.4.29

type CounterSet interface {
	// GetCounters gets a list and a map for consuming counter data
	GetCounters() (orderedKeys []CounterID, m map[CounterID]any)
	// ResetCounters resets all counters to their initial state
	ResetCounters(stopRateCounters bool)
}

CounterSet is the consumer interface for a counter set

type CounterSetData ¶ added in v0.4.41

type CounterSetData interface {
	Exists(name CounterID) (exists bool)
	// Value returns the monotonically increasing value for a possible plain counter
	//	- if no such counter exists, 0
	Value(name CounterID) (value uint64)
	// Running returns the increasing and decreasing running value for a possible plain counter
	//	- if no such counter exists, 0
	Get(name CounterID) (value, running, max uint64)
	// Rates returns the rate values a possible rate counter
	//	- if no such counter exists or values are not yet available, nil
	Rates(name CounterID) (rates map[RateType]float64)
	// DatapointValue returns the latest value a possible datapoint
	//	- if no such datapoint exists, 0
	DatapointValue(name CounterID) (value uint64)
	// DatapointMax returns the highest seen value for a possible datapoint
	//	- if no such datapoint exists, 0
	DatapointMax(name CounterID) (max uint64)
	// DatapointMin returns the lowest seen value for a possible datapoint
	//	- if no such datapoint exists, 0
	DatapointMin(name CounterID) (min uint64)
	// GetDatapoint returns dfatapoint data for a possible datapoint
	//	- if no such datapoint exists, 0
	GetDatapoint(name CounterID) (value, max, min uint64, isValid bool, average float64, n uint64)
}

CounterSetData provides simple access to a set of counters, rate counters and datapoints/

type CounterStore ¶ added in v0.4.100

type CounterStore interface {
	// GetOrCreateCounter retrieves a regular counter
	//	- never returns nil
	//	- type asserted to CounterValues
	GetCounter(name CounterID) (counter Counter)
	//GetCounter retrieves a counter that must exist
	//	- may return nil
	//	- type asserted to RateCounterValues or DatapointValue
	GetNamedCounter(name CounterID) (counter any)
}

CounterStore is a CounterSet consumer interface facilitating caching

type CounterValues ¶ added in v0.3.0

type CounterValues interface {
	// Get returns value/running/max with integrity. Thread-Safe
	//	- value is the monotonically increasing value
	//	- running is the fluctuating running value
	//	- max is the highest value running has had
	Get() (value, running, max uint64)
	// GetReset returns value/running/max with integrity and resets the counter. Thread-Safe
	//	- value is the monotonically increasing value
	//	- running is the fluctuating running value
	//	- max is the highest value running has had
	GetReset() (value, running, max uint64)
	// Value returns the monotonically increasing value. Thread-Safe
	//	- number of Inc invocations and positive Adds
	Value() (value uint64)
	// Running returns the fluctuating running value. Thread-Safe
	//	- number of Inc less Dec invocations and sum of Adds
	//	- never below 0
	Running() (running uint64)
	// Max returns the highest value running has had. Thread-Safe
	Max() (max uint64)
}

CounterValues is the consumer interface for a counter. Thread-safe

type Counters ¶ added in v0.3.0

type Counters interface {
	// GetOrCreateCounter is used by the data provider of a counter.
	//	- A counter has Inc and Dec operations.
	//	- Counters can be used to track number of currently operating invocations.
	//	- if period is present and positive, the counter created is a RateCounter.
	// Rate counters use threads, one per distinct period requested.
	// Rate counter threads are monitored by the provided Go group and failure
	// may force a counter restart or cause application termination. Failures
	// are typically caused by panics in the counter update task.
	//	- Counter is the data-provider side of a counter
	//	- CounterValues is the consumer side of a counter
	GetOrCreateCounter(name CounterID, period ...time.Duration) (counter Counter)
	// GetOrCreateCounter is used by the data provider of a datapoint.
	// A datapoint supports SetValue operation.
	// A datapoint tracks a quantity such as a latency value.
	GetOrCreateDatapoint(name CounterID, period time.Duration) (datapoint Datapoint)
}

Counters is the data provider interface for a counter set.

• max and running values are offered
• Counters and datapointy are thread-safe
• counters may be used to determine that code abide by intended paralellism and identifying hangs or abnormalisms.
• Printing counters every second can verify adequate progress and possibly identify blocking of threads or swapping and garbage collection outages.

type CountersFactory ¶ added in v0.3.0

type CountersFactory interface {
	// NewCounters returns a counter container.
	// is useCounters is false, the container does not actually do any counting.
	NewCounters(useCounters bool, g0 GoGen) (counters Counters)
}

CountersFactory is an abstract counter factory. CountersFactory enables providing of different counter implementations.

type CountingAwaitable ¶ added in v0.4.125

type CountingAwaitable struct {
	Awaitable
	// contains filtered or unexported fields
}

func NewCountingAwaitable() (awaitable *CountingAwaitable)

func (a *CountingAwaitable) Add(delta int)

func (a *CountingAwaitable) Count(delta ...int) (counter, adds int)

func (a *CountingAwaitable) Done()

func (a *CountingAwaitable) IsTriggered() (isTriggered bool)

type CyclicAwaitable ¶ added in v0.4.115

type CyclicAwaitable struct {
	// contains filtered or unexported fields
}

CyclicAwaitable is an awaitable that can be re-initialized

• one-to-many, happens-before
• the synchronization mechanic is closing channel, allowing consumers to await multiple events
• CyclicAwaitable.IsClosed provides thread-safe observability
• CyclicAwaitable.Close is idempotent, thread-safe and deferrable
• Open means event is pending, Close means event has triggered
• CyclicAwaitable.Open arms the awaitable returning a channel guaranteed to be open at timeof invocation
• —
• because Awaitable is renewed, access is via atomic Pointer
• Pointer to struct allows for atomic update of IsClosed and Open

Usage:

valueWaiter *CyclicAwaitable
…
func (v *V) getOrWaitForValue(value T) {
  var hasValue bool
  // check if value is already present
  if value, hasValue = v.hasValueFromThread(); hasValue {
    return
  }
  // arm cyclable
  v.valueWaiter.Open()
  // collect any value arriving prior to arming cyclable
  if value, hasValue = v.hasValueFromThread(); hasValue {
    return
  }
  <- v.valueWaiter.Ch()
  value, _ = v.hasValueFromThread()
…
func (v *V) threadStoresValue(value T) {
  v.store(value)
  v.valueWaiter.Close()

func (a *CyclicAwaitable) Ch() (ch AwaitableCh)

func (a *CyclicAwaitable) Close() (didClose bool)

func (a *CyclicAwaitable) IsClosed() (isClosed bool)

func (a *CyclicAwaitable) Open() (didOpen bool, ch AwaitableCh)

type DA ¶ added in v0.4.117

type DA *pruntime.CodeLocation

DA is the value returned by a deferred code location function

func A() DA

type DB ¶ added in v0.4.12

type DB interface {
	// Exec executes a query not returning any rows
	//	- ExecResult contains last inserted ID if any and rows affected
	Exec(partition DBPartition, query string, ctx context.Context,
		args ...any) (execResult ExecResult, err error)
	// Query executes a query returning zero or more rows
	Query(partition DBPartition, query string, ctx context.Context,
		args ...any) (sqlRows *sql.Rows, err error)
	// Query executes a query known to return exactly one row
	//	- zero rows returns error: sql: no rows in result set
	QueryRow(partition DBPartition, query string, ctx context.Context,
		args ...any) (sqlRow *sql.Row, err error)
	// Query executes a query known to return exactly one row and first column a string value
	QueryString(partition DBPartition, query string, ctx context.Context,
		args ...any) (value string, err error)
	// Query executes a query known to return exactly one row and first column an int value
	QueryInt(partition DBPartition, query string, ctx context.Context,
		args ...any) (value int, err error)
	// Close closes the database connection
	Close() (err error)
}

DB is a parallel database connection

• DB applies to any database implementation
• psql provides implementation with caching of:
• — DB objects and
• — prepared statements
• DB is obtained via new function like [DBFactory.NewDB]. Such returned DB can use:
• — its data-source namer to handle partitioning
• — delegation to its underlying possibly partitioned DB implementation
• — caching of DB implementation-objects and prepared statements
• — its schema function to bootstrap and migrate databases

type DBFactory ¶ added in v0.4.12

type DBFactory interface {
	// NewDB returns a DB object implementation.
	// The schema function executes application-specific SQL initialization for
	// a new datasource
	//	- executes CREATE of tables and indexes
	//	- configures database-specific referential integrity and journaling
	NewDB(
		dsnr DataSourceNamer,
		schema func(dataSource DataSource, ctx context.Context) (err error),
	) (db DB)
}

DBFactory is a standardized way to obtain DB objects

• DBFactory applies to any database implementation

type DBPartition ¶ added in v0.4.14

type DBPartition string

DBPartition is partition reference for a partitioned database

• partition is typically one table per year
• DBPartition applies to any database implementation

const NoPartition DBPartition = ""

type DSNrFactory ¶ added in v0.4.14

type DSNrFactory interface {
	// NewDSNr returns an object that can
	//	- provide data source names from partition selectors and
	//	- provide data sources from a data source name
	DataSourceNamer(appName string) (dsnr DataSourceNamer, err error)
}

DSNrFactory describes the signature for a data source namer new function

• the data source namer provides data sources for query operations
• DSNrFactory applies to any database implementation

type DataSource ¶ added in v0.4.12

type DataSource interface {
	// PrepareContext prepares a statement that is a query for reading or writing data
	//	- a prepared statement can be executed multiple times
	PrepareContext(ctx context.Context, query string) (stmt *sql.Stmt, err error)
	// Close closes the data-source
	Close() (err error)
}

DataSource represents a set of SQL tables, possibly partitioned, against which queries can be prepared and executed

• DataSource applies to any database implementation

type DataSourceName ¶ added in v0.4.127

type DataSourceName string

DataSource is a value referring to a set of SQL tables, possibly a partition

type DataSourceNamer ¶ added in v0.4.12

type DataSourceNamer interface {
	// DSN returns the data source name based on a partition selector
	DSN(partition ...DBPartition) (dataSourceName DataSourceName)
	// DataSource returns a usable data source based on a data source name
	DataSource(dsn DataSourceName) (dataSource DataSource, err error)
}

DataSourceNamer provides data source names for SQL based on possibly appplication name and partition year

• a data source represents a set of SQL tables, possibly partitioned, against which queries can be prepared and executed
• DataSourceNamer applies to any database implementation
• sqliter provides implementations for SQLite3

type Datapoint ¶ added in v0.4.29

type Datapoint interface {
	// SetValue records a value at time.Now().
	// SetValue supports method chaining.
	SetValue(value uint64) (datapoint Datapoint)
}

Datapoint tracks a value with average, max-min, and increase/decrease rates.

type DatapointValue ¶ added in v0.4.29

type DatapointValue interface {
	CloneDatapoint() (datapoint Datapoint)      // Clone takes a snapshot of a counter state.
	CloneDatapointReset() (datapoint Datapoint) // CloneReset takes a snapshot of a counter state and resets it to its initial state.
	GetDatapoint() (value, max, min uint64, isValid bool, average float64, n uint64)
	DatapointValue() (value uint64)
	DatapointMax() (max uint64)
	DatapointMin() (min uint64)
}

type Debouncer ¶ added in v0.4.26

type Debouncer[T any] struct {
	// contains filtered or unexported fields
}

Debouncer debounces event stream values

• T values are received from the in channel
• Once d time has elapsed with no further incoming Ts, a slice of read T values are provided to the sender function
• errFn receives any panics in the threads, expected none
• sender and errFn functions must be thread-safe.
• Debouncer is shutdown gracefully by closing the input channel or immediately by invoking the Shutdown method
• —
• two threads are launched per debouncer

func NewDebouncer[T any](
	debounceInterval, maxDelay time.Duration,
	inputCh <-chan T,
	sender func([]T),
	errFn AddError,
) (debouncer *Debouncer[T])

func (d *Debouncer[T]) Shutdown()

func (d *Debouncer[T]) Wait()

type Dent ¶ added in v0.3.0

type Dent interface {
	// Name is utf-8 base path in device file system.
	// Name is base name, ie. file name and extension.
	Name() (name string)
	// Modified time, the time file contents was changed, second precision, continuous time
	Modified() (modified time.Time)
	// IsDir indicates directory.
	// LIST only support symbolic link, directory and regular file types
	IsDir() (isDir bool)
	// IsRegular indicates regular file, ie. not a directory or symbolic link.
	// LIST only support symbolic link, directory and regular file types
	IsRegular() (isRegular bool) // ie.not directory or symlink
	// Perm returns os.FileMode data.
	// 9-bit Unix permissions per os.FilePerm.
	// LIST also supports directory and symlink bits
	Perm() (perm fs.FileMode)
	// Size is limited to 4 GiB-1
	Size() (size uint32)
}

Dent is the information returned by adb ls or LIST

type Devicette ¶ added in v0.3.0

type Devicette interface {
	// Serial returns the serial number for this device
	Serial() (serial AndroidSerial)
	// Shell executes a shell command on the device.
	//	- the response is a byte sequence
	//	- note: interning large strings may lead to memory leaks
	Shell(command string) (out []byte, err error)
	// ShellStream executes a shell command on the device returning a readable socket
	ShellStream(command string) (conn io.ReadWriteCloser, err error)
	/*
		Pull copies a remote file or directory on the Android device to a local file system location.
		the local file must not exist.
		Pull refuses certain files like product apks. shell cat works
	*/
	Pull(remotePath, nearPath string) (err error)
	/*
		List has some peculiarities:
		If remoteDir is not an existing directory, an empty list is returned.
		Entries with insufficient permisions are ignored.
		Update: . and .. are removed, adb LIST ortherwise do return those.
		File mode: the only present type bits beyond 9-bit Unix permissions are
		symlink, regular file and directory.
		File size is limited to 4 GiB-1.
		Modification time resolution is second and range is confined to a 32-bit Unix timestamp.
	*/
	List(remoteDir string) (dFileInfo []Dent, err error)
}

Devicette is a generic implementation of the capabilities of a device implementing the adb Android debug bridge protocol

type DevicetteFactory ¶ added in v0.4.0

type DevicetteFactory interface {
	// NewDevicette creates a Devicette interacting with remote adb Android Debug Bridge
	// devices via an adb server available at the socket address address
	NewDevicette(address AdbSocketAddress, serial AndroidSerial, ctx context.Context) (devicette Devicette)
}

DevicetteFactory describes how Devicette objects are obtained.

type DoErr ¶ added in v0.4.26

type DoErr interface {
	DoErr(op func() (err error)) (err error)
}

type Doneable ¶ added in v0.4.12

type Doneable interface {
	Add(delta int)
	Done()
}

Doneable is the callee part of sync.Waitgroup and other implementations Doneable is a many-to-many relation. Doneable allows the callee to instatiate and invoke any number of things that are awaitable by the caller.

… = NewSomething(&waitsForLots, &shutsDownLots)
go someThread(&waitsForLots, &shutsDownLots)
func someThread(Doneable w, context.Context ctx) {
  defer w.Done()
  w.Add(2)
  go somethingElse()

type EchoModerator ¶ added in v0.4.41

type EchoModerator struct {
	// contains filtered or unexported fields
}

EchoModerator is a parallelism-limiting Moderator that:

• prints any increase in parallelism over the concurrency value
• prints exhibited invocation slowness exceeding latencyWarningPoint
• prints progressive slowness exceeding latencyWarningPoint for non-returning invocations in progress on schedule timerPeriod
• EchoModerator can be used in production for ongoing diagnose of apis and libraries
• cost is one thread, one timer, and a locked linked-list of invocations
• —
• EchoModerator is intended to control and diagnose [exec.Command] invocations
• problems include:
• — too many parallel invocations
• — invocations that do not return or are long running
• — too many threads held waiting to invoke
• — unexpected behavior under load
• — deviating behavior when operated for extended periods of time

func NewEchoModerator(
	concurrency uint64,
	latencyWarningPoint time.Duration,
	waitingWarningPoint uint64,
	timerPeriod time.Duration,
	label string, goGen GoGen, log PrintfFunc,
) (echoModerator *EchoModerator)

func (m *EchoModerator) Ticket() (returnTicket func())

func moderatedFunc() {
  defer echoModerator.Ticket()()

type Enum ¶ added in v0.4.33

type Enum[T any] interface {
	KeyEnum[string, T]
}

Enum is an enumeration using string keys mapping to a value type T. T is a unique named type that separates the values of this enumeration from all other values.

type EnumItem ¶ added in v0.4.33

type EnumItem[K constraints.Ordered, V any] interface {
	// Key returns the key for this enumeration value
	Key() (key K)
	// Description returns a descriptive sentence for this enumeration value
	Description() (desc string)
	// Value returns this enumeration value’s value using the restricted type
	Value() (value V)

	fmt.Stringer
}

EnumItem is a generic interface for enumeration item implementations. Enumeration items are ordered by the K key type.

• K is a key type whose values map to restricted type V values one-to-one.
• V is a restricted type for enumeration values that may store more efficiently compared to a portable type.

type ErrorCloser ¶ added in v0.4.20

type ErrorCloser interface {
	InvokeIfError(addError func(err error))
	Close()
}

type ErrorManager ¶ added in v0.4.20

type ErrorManager interface {
	Ch() (ch <-chan GoError)
}

type ErrorSink ¶ added in v0.4.20

type ErrorSink interface {
	AddError(err error)
}

type ExecResult ¶ added in v0.4.14

type ExecResult interface {
	// - ID is last inserted ID if any
	// - rows is number of rows affected
	Get() (ID int64, rows int64)
	// “sql.Result: ID afe3… rows: 123”
	String() (s string)
}

ExecResult is the result from [DB.Exec], a query not returning rows

type FSLocation ¶

type FSLocation interface {
	Directory() (directory string)
}

type Frame ¶ added in v0.4.20

type Frame interface {
	// the code location for this frame, never nil
	Loc() (location *pruntime.CodeLocation)
	// function argument values like “(1, 0x14000113040)”
	//	- values of basic types like int are displayed
	//	- most types appear as a pointer value “0x…”
	Args() (args string)
	// prints the Frame suitable to be part of a stack trace
	//   - fully qualified package name with function or type and method
	//     and argument values
	//   - absolute path to source file and line number
	//
	// output:
	//
	//	github.com/haraldrudell/parl/pdebug.TestFrame(0x1400014a340)␤
	//	␠␠frame_test.go:15
	String() (s string)
}

Frame represents an executing code location, ie. a code line in source code

• parl.Frame is similar to runtime.Frame returned by runtime.CallersFrames but has only basic types, ie. it can be printed, stored and transferred

type Future ¶ added in v0.4.26

type Future[T any] struct {
	// contains filtered or unexported fields
}

Future is a container for an awaitable calculation result

• Future allows a thread to await a value calculated in parallel by other threads
• unlike for a promise, consumer manages any thread, therefore producing debuggable code and meaningful stack traces
• a promise launches the thread why there is no trace of what code created the promise or why

func NewFuture[T any]() (calculation *Future[T])

 var calculation = NewFuture[someType]()
 go calculateThread(calculation)
 …
 var result, isValid = calculation.Result()

func calculateThread(future *Future[someType]) {
  var err error
  var isPanic bool
  var value someType
  defer calculation.End(&value, &isPanic, &err)
  defer parl.RecoverErr(func() parl.DA { return parl.A() }, &err, &isPanic)

   value = …

func (f *Future[T]) Ch() (ch AwaitableCh)

func (f *Future[T]) End(value *T, isPanic *bool, errp *error)

func (f *Future[T]) IsCompleted() (isCompleted bool)

func (f *Future[T]) Result() (result T, hasValue bool)

func (f *Future[T]) TResult() (tResult *TResult[T])

type Go ¶ added in v0.4.12

type Go interface {
	// Register performs no function but allows the Go object to collect
	// information on the new thread.
	// - label is an optional name that can be assigned to a Go goroutine thread
	Register(label ...string) (g0 Go)
	// AddError emits a non-fatal errors
	AddError(err error)
	// Go returns a Go object to be provided as a go-statement function-argument
	//	in a function call invocation launching a new gorotuine thread.
	//	- the new thread belongs to the same GoGroup thread-group as the Go
	//		object whose Go method was invoked.
	Go() (g0 Go)
	// SubGo returns a GoGroup thread-group whose fatal and non-fatel errors go to
	//	the Go object’s parent thread-group.
	//	- a SubGo is used to ensure sub-threads exiting prior to their parent thread
	//		or to facilitate separate cancelation of the threads in the subordinate thread-group.
	//	- fatal errors from SubGo threads are handled in the same way as those of the
	//		Go object, typically terminating the application.
	//   - the SubGo thread-group terminates when both its own threads have exited and
	//	- the threads of its subordinate thread-groups.
	SubGo(onFirstFatal ...GoFatalCallback) (subGo SubGo)
	// SubGroup returns a thread-group with its own error channel.
	//	- a SubGroup is used for threads whose fatal errors should be handled
	//		in the Go thread.
	//	- The threads of the Subgroup can be canceled separately.
	//		- SubGroup’s error channel collects fatal thread terminations
	//		- the SubGroup’s error channel needs to be read in real-time or after
	//		SubGroup termination
	//   - non-fatal errors in SubGroup threads are sent to the Go object’s parent
	//		similar to the AddError method
	//   - the SubGroup thread-group terminates when both its own threads have exited and
	//	- the threads of its subordinate thread-groups.
	SubGroup(onFirstFatal ...GoFatalCallback) (subGroup SubGroup)
	// Done indicates that this goroutine is exiting
	//	- err == nil means successful exit
	//	- non-nil err indicates fatal error
	// 	- deferrable
	Done(errp *error)
	// Wait awaits exit of this Go thread.
	Wait()
	WaitCh() (ch AwaitableCh)
	// Cancel signals for the threads in this Go thread’s parent GoGroup thread-group
	// and any subordinate thread-groups to exit.
	Cancel()
	// Context will Cancel when the parent thread-group Cancels
	//	or Cancel is invoked on this Go object.
	//	- Subordinate thread-groups do not Cancel the context of the Go thread.
	Context() (ctx context.Context)
	// ThreadInfo returns thread data that is partially or fully populated
	//	- ThreadID may be invalid: threadID.IsValid.
	//	- goFunction may be zero-value: goFunction.IsSet
	//	- those values present after public methods of parl.Go has been invoked by
	//		the new goroutine
	ThreadInfo() (threadData ThreadData)
	// values always present
	Creator() (threadID ThreadID, createLocation *pruntime.CodeLocation)
	//	- ThreadID may be invalid: threadID.IsValid.
	//	- goFunction may be zero-value: goFunction.IsSet
	//	- those values present after public methods of parl.Go has been invoked by
	//		the new goroutine
	GoRoutine() (threadID ThreadID, goFunction *pruntime.CodeLocation)
	// GoID efficiently returns the goroutine ID that may be invalid
	//	- valid after public methods of parl.Go has been invoked by
	//		the new goroutine
	GoID() (threadID ThreadID)
	// EntityID returns a value unique for this Go
	//	- ordered: usable as map key or for sorting
	//	- always valid, has .String method
	EntityID() (goEntityID GoEntityID)
	fmt.Stringer
}

Go provides the four needs of a running goroutione thread. The Go is provided as a function argument in the go statement function call that launches the thread.

• the four needs:
• — to be waited upon via [Go.Done]
• — to submit non-fatal errors via [Go.AddError]
• — to detect and initiate cancel via [Go.Context] [Go.Cancel]
• [Go.Cancel] cancels:
• — this Go thread’s context
• — this Go’s parent thread-group’s context and
• — this Go’s parent thread-group’s subordinate thread-groups’ contexts
• The Go Context is canceled when
• — the parent GoGroup thread-group’s context is Canceled or
• —a thread in the parent GoGroup thread-group initiates Cancel
• Cancel by threads in sub ordinate thread-groups do not Cancel this Go thread

Usage:

var threadGroup = g0.NewGoGroup(context.Background())
go someFunc(threadGroup.Go())
…
func someFunc(g parl.Go) {
  var err error
  defer g.Register().Done(&err)
  defer parl.RecoverErr(func() parl.DA { return parl.A() }, &err)
  …

type GoDebug ¶ added in v0.4.71

type GoDebug uint8

const (
	NoDebug GoDebug = iota
	DebugPrint
	AggregateThread
)

type GoEntityID ¶ added in v0.4.117

type GoEntityID uint64

GoEntityID is a unique named type for Go objects

• GoEntityID is required becaue for Go objects, the thread ID is not available prior to the go statement and GoGroups do not have any other unique ID
• GoEntityID is suitable as a map key
• GoEntityID uniquely identifies any Go-thread GoGroup, SubGo or SubGroup

func (i GoEntityID) String() (s string)

type GoError ¶ added in v0.4.12

type GoError interface {
	error // Error() string
	// Err retrieves the original error value
	Err() (err error)
	// ErrString returns string representation of error
	//   - if no error “OK”
	//   - if not debug or panic, short error with location
	//   - otherwise error with stack trace
	ErrString() (errString string)
	// Time provides when this error occurred
	Time() time.Time
	// IsThreadExit determines if this error is a thread exit, ie. GeExit GePreDoneExit
	//	- thread exits may have err nil
	//	- fatals are non-nil thread exits that may require specific actions such as
	//		application termination
	IsThreadExit() (isThreadExit bool)
	// IsFatal determines if this error is a fatal thread-exit, ie. a thread exiting with non-nil error
	IsFatal() (isThreadExit bool)
	// ErrContext returns in what situation this error occurred
	ErrContext() (errContext GoErrorContext)
	// Go provides the thread and goroutine emitting this error
	Go() (g0 Go)
	fmt.Stringer
}

GoError is an error or a thread exit associated with a goroutine

• GoError encapsulates the original unadulterated error
• GoError provides context for taking action on the error
• Go provides the thread associated with the error. All GoErrors are associated with a Go object
• because GoError is both error and fmt.Stringer, to print its string representation requires using the String() method, otherwise fmt.Printf defaults to the Error() method

type GoErrorContext ¶ added in v0.4.29

type GoErrorContext uint8

const (
	// GeNonFatal indicates a non-fatal error ocurring during processing.
	// err is non-nil
	GeNonFatal GoErrorContext = iota + 1
	// GePreDoneExit indicates an exit value of a subordinate goroutine,
	// other than the final exit of the last running goroutine.
	// err may be nil
	GePreDoneExit
	// A SubGroup with its own error channel is sending a
	// locally fatal error not intended to terminate the app
	GeLocalChan
	// A thread is requesting app termination without a fatal error.
	//	- this could be a callback
	GeTerminate
	// GeExit indicates exit of the last goroutine.
	// err may be nil.
	// The error channel may close after GeExit.
	GeExit
)

func (ge GoErrorContext) String() (s string)

type GoFactory ¶ added in v0.4.29

type GoFactory interface {
	// NewGo returns a light-weight thread-group.
	//	- GoGroup only receives Cancel from ctx, it does not cancel this context.
	NewGoGroup(ctx context.Context, onFirstFatal ...GoFatalCallback) (g0 GoGroup)
}

type GoFatalCallback ¶ added in v0.4.29

type GoFatalCallback func(goGen GoGen)

GoFatalCallback receives the thread-group on its first fatal thread-exit

• GoFatalCallback is an optional onFirstFatal argument to
• — NewGoGroup
• — SubGo
• — SubGroup

type GoGen ¶ added in v0.4.29

type GoGen interface {
	// Go returns a Go object to be provided as a go statement function argument.
	Go() (g0 Go)
	// SubGo returns a thread-group whose fatal errors go to GoGen’s parent.
	//   - both non-fatal and fatal errors in SubGo threads are sent to GoGen’s parent
	// 		like Go.AddError and Go.Done.
	//		- therefore, when a SubGo thread fails, the application will typically exit.
	//		- by awaiting SubGo, Go can delay its exit until SubGo has terminated
	//   - the SubGo thread-group terminates when the its thread exits
	SubGo(onFirstFatal ...GoFatalCallback) (subGo SubGo)
	// SubGroup creates a sub-ordinate thread-group
	SubGroup(onFirstFatal ...GoFatalCallback) (subGroup SubGroup)
	// Cancel terminates the threads in the Go consumer thread-group.
	Cancel()
	// Context will Cancel when the parent thread-group Cancels.
	// Subordinate thread-groups do not Cancel this context.
	Context() (ctx context.Context)
}

GoGen allows for new Go threads, new SubGo and SubGroup thread-groups and cancel of threads in the thread-group and its subordinate thread-groups.

• GoGen is value from NewGoGroup GoGroup SubGo SubGroup Go, ie. any Go-interface object

type GoGroup ¶ added in v0.4.18

type GoGroup interface {
	// Go returns a Go object to be provided as a go statement function argument.
	Go() (g0 Go)
	// SubGo returns athread-group whose fatal errors go to Go’s parent.
	//   - both non-fatal and fatal errors in SubGo threads are sent to Go’s parent
	// 		like Go.AddError and Go.Done.
	//		- therefore, when a SubGo thread fails, the application will typically exit.
	//		- by awaiting SubGo, Go can delay its exit until SubGo has terminated
	//   - the SubGo thread-group terminates when the its thread exits
	SubGo(onFirstFatal ...GoFatalCallback) (subGo SubGo)
	// SubGroup creates a sub-ordinate GoGroup.
	//	- SubGroup fatal and non-fatal errors are sent to the parent GoGroup.
	//	-	SubGroup-context initiated Cancel only affect threads in the SubGroup thread-group
	//	- parent-initiated Cancel terminates SubGroup threads
	//	- SubGroup only awaits SubGroup threads
	//	- parent await also awaits SubGroup threads
	SubGroup(onFirstFatal ...GoFatalCallback) (subGroup SubGroup)
	// Ch returns a channel sending the all fatal termination errors when
	// the FailChannel option is present, or only the first when both
	// FailChannel and StoreSubsequentFail options are present.
	Ch() (ch <-chan GoError)
	// Wait waits for this thread-group to terminate.
	Wait()
	// EnableTermination controls temporarily preventing the GoGroup from
	// terminating.
	// EnableTermination is initially true.
	//	- invoked with no argument returns the current state of EnableTermination
	//	- invoked with [AllowTermination] again allows for termination and
	//		immediately terminates the thread-group if no threads are currently running.
	//	- invoked with [PreventTermination] allows for the number of managed
	//		threads to be temporarily zero without terminating the thread-group
	EnableTermination(allowTermination ...bool) (mayTerminate bool)
	// Cancel terminates the threads in this and subordinate thread-groups.
	Cancel()
	// Context will Cancel when the parent context Cancels.
	// Subordinate thread-groups do not Cancel this context.
	Context() (ctx context.Context)
	// the available data for all threads
	Threads() (threads []ThreadData)
	// threads that have been named ordered by name
	NamedThreads() (threads []ThreadData)
	// SetDebug enables debug logging on this particular instance
	//	- parl.NoDebug
	//	- parl.DebugPrint
	//	- parl.AggregateThread
	SetDebug(debug GoDebug, log ...PrintfFunc)
	fmt.Stringer
}

GoGroup manages a hierarchy of threads

• GoGroup only terminates when:
• — the last thread in its hierarchy exits
• — [GoGroup.EnableTermination] is set to true when no Go threads exist in the hierarchy
• the GoGroup hierarchy consists of:
• — managed goroutines returned by [GoGroup.Go]
• — a SubGo subordinate thread-group hierarchy returned by [GoGroup.SubGo] that allows for a group of threads to be canceled or waited upon separately
• — a SubGroup subordinate thread-group hierarchy returned by [GoGroup.SubGroup] that allows for a group of threads to exit with fatal errors without canceling the GoGroup and for those threads to be canceled or waited upon separately
• — each subordinate Go thread or SubGo or SubGroup subordinate thread-groups can create additional threads and subordinate thread-groups.
• [GoGroup.Context] returns a context that is the context or parent context of all the Go threads, SubGo and SubGroup subordinate thread-groups in its hierarchy
• [GoGroup.Cancel] cancels the GoGroup Context, thereby signaling to all threads in the GoGroup hierarchy to exit. This will eventually terminate the GoGroup
• providing a parent context to the GoGroup implementation allows for terminating the GoGroup via this parent context
• A thread invoking [Go.Cancel] will signal to all threads in its GoGroup or SubGo or SubGroup thread-groups to exit. It will also signal to all threads in its subordinate thread-groups to exit. This will eventually terminate its threadgroup and all that threadgroup’s subordinate threadgroups.
• Alternatives to parl.Go is parl.NewGoResult and parl.NewGoResult2 that only provides being awaitable to a goroutine
• —
• from this thread-group will terminate all threads in this and subordinate thread-groups if this thread-group was provided the FirstFailTerminates option, which is default.
• A fatal thread-termination in a sub thread-group only affects this thread-group if the sub thread-group was provided a nil fatal function, the FirstFailTerminates option, which is default, and no explicit FailChannel option.
• Fatal thread terminations will propagate to parent thread-groups if this thread group did not have a fatal function provided and was not explicitly provided the FailChannel option.
• A Cancel in this thread-group or in a parent context cancels threads in this and all subordinate thread-groups.
• A Cancel in a subordinate thread-group does not affect this thread-group.
• Wait in this thread-group wait for threads in this and all subordinate thread-groups.

type GoResult ¶ added in v0.4.134

type GoResult struct {
	// contains filtered or unexported fields
}

GoResult makes any number of goroutines awaitable

• number of goroutines must be known at time of new
• NewGoResult is the simplest, goroutines are awaited by [GoResult.ReceiveError]
• NewGoResult2 also has [IsError] method indicating if any goroutine exited with fatal error
• GoResult.IsValid true if the GoResult is initialized
• [GoResult.SendError](errp *error) deferrable, how goroutine sends results
• [GoResult.ReceiveError](errp *error, n ...int) (err error)
• [GoResult.Count]() (count int) number of buffered errors
• NewGoResult2 also has:
• — [GoResult.IsError]() (isError bool) true if any goroutine returned error
• — [GoResult.SetIsError]() sets the error flag manually
• — [GoResult.Remaining]() (remaining int) number of goroutines that have yet to exit
• —
• passed by value
• getting around that receiver cannot be interface
• receiver is value struct with pointer in the form of an interface

func NewGoResult(n ...int) (goResult GoResult)

func someFunc(text string) (err error) {
  var g = parl.NewGoResult()
  go goroutine(text, g)
  defer g.ReceiveError(&err)
  …
func goroutine(text string, g parl.GoResult) {
  var err error
  defer g.SendError(&err)
  defer parl.RecoverErr(func() parl.DA { return parl.A() }, &err)

  err = …

func NewGoResult2(n ...int) (goResult GoResult)

func (g GoResult) IsValid() (isValid bool)

type IdempotentCloser ¶ added in v0.4.166

type IdempotentCloser[C io.Closer] struct {
	// contains filtered or unexported fields
}

func NewIdemPotentCloser[C io.Closer](closer C) (idempotentCloser *IdempotentCloser[C])

func (c *IdempotentCloser[C]) Close() (err error)

func (c *IdempotentCloser[C]) IsClose() (isClose bool)

type Interval ¶ added in v0.4.48

type Interval struct {
	// contains filtered or unexported fields
}

type Invocation ¶ added in v0.4.130

type Invocation[T any] struct {
	Prev, Next atomic.Pointer[Invocation[T]]
	ThreadID   ThreadID
	Value      T
	// contains filtered or unexported fields
}

a timed invocation created by [InvokeTimer.Invocation]

• holds invocation instance-data for [InvocationTimer.invocationEnd]

func NewInvocation[T any](invocationEnd func(invocation *Invocation[T], duration time.Duration), value T) (invocation *Invocation[T])

func (i *Invocation[T]) Age() (age time.Duration)

func (i *Invocation[T]) DeferFunc()

type InvocationTimer ¶ added in v0.4.130

type InvocationTimer[T any] struct {
	// contains filtered or unexported fields
}

InvocationTimer monitors funtion invocations for parallelism and latency

• callback is invoked on exceeding thresholds and reaching a new max
• runs one thread per instance while an invocation is active

func NewInvocationTimer[T any](
	callback CBFunc, endCb func(T),
	latencyWarningPoint time.Duration, parallelismWarningPoint uint64,
	timerPeriod time.Duration,
	goGen GoGen,
) (invokeTimer *InvocationTimer[T])

func (i *InvocationTimer[T]) Invocation(value T) (deferFunc func())

func someFunc() {
  defer invocationTimer.Invocation()()

func (i *InvocationTimer[T]) Oldest() (age time.Duration, threadID ThreadID)

type KeyEnum ¶ added in v0.4.33

type KeyEnum[K constraints.Ordered, T any] interface {
	IsKey(key K) (isKey bool)                  // IsKey checks whether key maps to an enumerated value
	Value(key K) (enum T, err error)           // Value looks up an enumerated value by key
	KeyIterator() (iterator iters.Iterator[K]) // KeyIterator returns an iterator that iterates over all keys in order of definition

	IsValid(enum T) (isEnumValue bool)      // IsValid checks whether value is among enumerated values
	Key(value T) (key K, err error)         // Key gets the key value for an enumerated T value
	ValueAny(value any) (enum T, err error) // ValueAny attempts to convert any value to a T enumerated value
	Iterator() (iterator iters.Iterator[T]) // Iterator returns an iterator that iterates over all enumerated values in order of definition
	Description(enum T) (desc string)       // Description gets a descriptive sentence for an enum value
	StringT(enum T) (s string)              // StringT provides a string representation for an enumeration value
	// Compare compares two T values.
	//	- result is 0 if the two values are considered equal
	//	- result is 1 if value1 is considered greater than value2
	//	- result is -1 if value1 is considered less than value2
	Compare(value1, value2 T) (result int)

	Name() (s string) // Name returns a short string naming this enumeration
	fmt.Stringer
}

KeyedEnum is an enumeration using a key type K mapping to a value type T.

• T is a unique named type that separates the values of this enumeration from all other values.
• K is a common type, often a single-word string, whose values are unique and maps one-to-one to a T value. K is constraints.Ordered and can be used as a map key.
• The implementation’s stored type may be different from both K and T.

Some benefits with enumerations are:

• unknown, illegal or value duplications are detected
• integral values and their meanings can be printed
• one set of integral values are not confused with another, eg. unix.AF_INET and unix.RTAX_BRD
• Reveals the meaning when used as function arguments and other allowed values can be examined

type Moderator ¶

type Moderator struct {
	// contains filtered or unexported fields
}

Moderator invokes functions at a limited level of parallelism. It is a ticketing system

m := NewModerator(20, ctx)
m.Do(func() (err error) { // waiting here for a ticket
  // got a ticket!
  …
  return or panic // ticket automatically returned
m.String() → waiting: 2(20)

type ModeratorCore ¶ added in v0.4.26

type ModeratorCore struct {
	// contains filtered or unexported fields
}

ModeratorCore invokes functions at a limited level of parallelism

• ModeratorCore is a ticketing system
• ModeratorCore does not have a cancel feature
• during low contention atomic performance
• during high-contention lock performance

Usage:

m := NewModeratorCore(20, ctx)
defer m.Ticket()() // waiting here for a ticket
// got a ticket!
…
return or panic // ticket automatically returned
m.String() → waiting: 2(20)

func NewModeratorCore(parallelism uint64) (m *ModeratorCore)

func (m *ModeratorCore) Status() (parallelism, active, waiting uint64)

func (m *ModeratorCore) String() (s string)

func (m *ModeratorCore) Ticket() (returnTicket func())

defer moderator.Ticket()()

type MutexWait ¶ added in v0.4.54

type MutexWait struct {
	// contains filtered or unexported fields
}

MutexWait is maximum-lightweight observable single-fire Mutex. Thread-Safe

func NewMutexWait() (mutexWait *MutexWait)

func (mw *MutexWait) IsUnlocked() (isUnlocked bool)

func (mw *MutexWait) Unlock()

func (mw *MutexWait) Wait()

type NBChan ¶ added in v0.4.0

type NBChan[T any] struct {
	perrors.ParlError // thread panics and channel close errors
	// contains filtered or unexported fields
}

NBChan is a non-blocking send channel with trillion-size queues.

• NBChan behaves both like a channel and a thread-safe slice
• — efficiency of sending and receiving multiple items at once
• — ability to wait for items to become available
• NBChan is initialization-free, thread-safe, idempotent and observable with panic-free methods and deferrable medthods
• values are sent to the channel using Send/SendMany that are never blocked by channel send and are panic-free error-free
• values are received from the channel or fetched all or many at once using Get
• — Get wait:DataWaitCh and NBChanNone is highest throughput at lowest cpu load
• — — cost is no channel is sending values, Get must be used
• — — benefit is no thread
• — Get Ch and NBChanAlways is higher throughput than regular thread
• — — cost is thread is always running
• — with regular thread or NBChanAlways:
• — — Ch offers wait with channel receive
• — — DataWaitCh only waits for data available
• NBChan has deferrable, panic-free, observable, idempotent close. The underlying channel is closed when:
• — Close is invoked and the channel is read to end
• — CloseNow is invoked
• NBChan is observable:
• — DidClose indicates whether Close or CloseNow has been invoked
• — IsClosed indicates whether the underlying channel has closed
• — WaitForClose is deferrable and panic-free and waits until the underlying channel has been closed.
• — WaitForCloseCh returns a channel that closes when the underlying channel closes
• NBChan is designed for error-free operation and only has panics and close errrors. All errors can be collected via:
• — CloseNow WaitForClose GetError
• NBChan has contention-separation between Send/SendMany and Get
• NBChan can be used as an error channel where the sending thread does not block from a delayed or missing reader.

Usage:

var errCh parl.NBChan[error]
go thread(&errCh)
err, ok := <-errCh.Ch()
errCh.WaitForClose()
errCh.GetError()
…
func thread(errCh *parl.NBChan[error]) {
defer errCh.Close() // non-blocking close effective on send complete
var err error
defer parl.Recover(parl."", &err, errCh.AddErrorProc)
errCh.Ch() <- err // non-blocking
if err = someFunc(); err != nil {
err = perrors.Errorf("someFunc: %w", err)
return

func NewNBChan[T any](threadType ...NBChanThreadType) (nbChan *NBChan[T])

var nbChan NBChan[error]
go thread(&nbChan)

func (n *NBChan[T]) Capacity() (capacity int)

func (n *NBChan[T]) Ch() (ch <-chan T)

func (n *NBChan[T]) Close() (didClose bool)

func (n *NBChan[T]) CloseNow(errp ...*error) (didClose bool, err error)

func (n *NBChan[T]) Count() (unsentCount int)

func (n *NBChan[T]) DataWaitCh() (ch <-chan struct{})

func (n *NBChan[T]) DidClose() (didClose bool)

func (n *NBChan[T]) Get(elementCount ...int) (allItems []T)

func (n *NBChan[T]) IsClosed() (isClosed bool)

func (n *NBChan[T]) Scavenge(setCapacity int)

func (n *NBChan[T]) Send(value T)

func (n *NBChan[T]) SendMany(values []T)

func (n *NBChan[T]) SetAllocationSize(size int) (nb *NBChan[T])

func (n *NBChan[T]) SetAlwaysThread() (nb *NBChan[T])

func (n *NBChan[T]) SetNoThread()

func (n *NBChan[T]) ThreadStatus() (threadStatus string)

func (n *NBChan[T]) WaitForClose(errp ...*error)

func (n *NBChan[T]) WaitForCloseCh() (ch <-chan struct{})

type NBChanThreadType ¶ added in v0.4.100

type NBChanThreadType uint8

const (
	// NBChanAlways configures NBChan to always have a thread
	//   - benefit: for empty NBChan, Send/SendMany to  channel receive is faster due to avoiding thread launch
	//   - cost: a thread is always running instad of only running when NBChan non-empty
	//   - cost: Close or CloseNow must be invoked to shutdown NBChan
	NBChanAlways NBChanThreadType = iota + 1
	// NBChanNone configures no thread.
	//	- benefit: lower cpu
	//	- cost: data can only be received using [NBChan.Get]
	NBChanNone
)

func (n NBChanThreadType) String() (s string)

type NotifierFunc ¶ added in v0.4.97

type NotifierFunc func(slice Stack)

a NotifierFunc receives a stack trace of function causing cancel

• typically stack trace begins with parl.InvokeCancel

type OnError ¶ added in v0.4.117

type OnError func(err error)

OnError is a function that receives error values from an errp error pointer or a panic

var NoOnError OnError

type OnErrorStrategy ¶ added in v0.4.135

type OnErrorStrategy uint8

how OnError is handled: recoverOnErrrorOnce recoverOnErrrorMultiple recoverOnErrrorNone

type Once ¶ added in v0.4.18

type Once struct {
	// contains filtered or unexported fields
}

parl.Once is an observable sync.Once with an alternative DoErr method

• Once.DoErr invokes a function returning error recovering a panic
• Once.IsDone returns whether the Once has been executed, atomic performance
• Once.Result returns a possible Once.DoErr outcome, atomic performance
• Once.Do is similar to sync.Once.Do
• parl.Once is thread-safe and does not require initialization
• No thread will return from Once.Do or Once.DoErr until once.Do or once.DoErr has completed

func (o *Once) Do(doFuncArgument func())

var once parl.Once
once.Do(myFunc)
…
if once.IsDone() …
func myFunc() { …

func (o *Once) DoErr(doErrFuncArgument func() (err error)) (didOnce, isPanic bool, err error)

var once parl.Once
var didTheClose, isPanic, err = once.DoErr(osFile.Close)

func (o *Once) IsDone() (isDone bool)

func (o *Once) Result() (isPanic bool, hasResult bool, err error)

type OnceCh ¶ added in v0.4.161

type OnceCh struct {
	// contains filtered or unexported fields
}

OnceCh implements a one-time execution filter

• initialization free

Usage:

var o OnceCh
if isWinner, closeFunc := o.IsWinner(); !isWinner {
  return // thread already waited for winner thread completion
} else {
  defer closeFunc()
}
…

func (o *OnceCh) Ch() (ch AwaitableCh)

func (o *OnceCh) IsClosed() (isClosed bool)

func (o *OnceCh) IsInvoked() (isInvoked bool)

func (o *OnceCh) IsWinner(noWait ...bool) (isWinner bool, closeFunc func())

type OnceWaiter ¶ added in v0.4.29

type OnceWaiter struct {
	// contains filtered or unexported fields
}

OnceWaiter allows any number of threads to wait for a single next occurrence.

• the occurrence is trigger in any of two ways:
• — a parent context may be passed in that on cancel triggers the wait
• — the Cancel method is invoked
• Ch returns a channel that sends one item on the occurrence but never closes
• Done returns a channel that closes on the occurrence happens, similar to a context
• Wait awaits the occurrence
• a did-occurer object can be obtained that returns true once the cycle trigs.
• a context can be obtained that cancels on the next trig
• the cycles can be permanently canceled or trigged and rearmed

func NewOnceWaiter(ctx context.Context) (onceReceiver *OnceWaiter)

func (ow *OnceWaiter) Cancel()

func (ow *OnceWaiter) Ch() (ch <-chan struct{})

func (cw *OnceWaiter) Context() (ctx context.Context)

func (ow *OnceWaiter) DidOccur() (didOccur bool)

func (ow *OnceWaiter) Done() (done <-chan struct{})

func (ow *OnceWaiter) Wait()

type OnceWaiterRO ¶ added in v0.4.29

type OnceWaiterRO struct {
	// contains filtered or unexported fields
}

OnceWaiterRO allows any number of threads to wait for a single next occurrence. OnceWaiterRO is a OnceWaiter without the Cancel method

func NewOnceWaiterRO(onceWaiter *OnceWaiter) (onceWaiterRO *OnceWaiterRO)

func (ow *OnceWaiterRO) Ch() (ch <-chan struct{})

func (ow *OnceWaiterRO) Context() (ctx context.Context)

func (ow *OnceWaiterRO) DidOccur() (didOccur bool)

func (ow *OnceWaiterRO) Done() (done <-chan struct{})

func (ow *OnceWaiterRO) Wait()

type Password ¶

type Password interface {
	// true if a password can be obtained interactively
	HasPassword() (hasPassword bool)
	// blocking interactive password input
	Password() (password []byte, err error)
}

Password indicates if a password can be read interactively

• [pterm.NoPassword] implements password input unavailable
• [pterm.NewPassword] implements interactive password input

type PemBytes ¶ added in v0.4.26

type PemBytes []byte

PemBytes bytes is 7-bit ascii string representing keys or certificates

type PeriodWaiter ¶ added in v0.4.100

type PeriodWaiter struct {
	// contains filtered or unexported fields
}

PeriodWaiter blocks Wait invokers while a HoldWaiters that has not been succeeded by a ReleaseWaiters invocation. Thread-safe

func NewPeriodWaiter() (periodWaiter *PeriodWaiter)

func (p *PeriodWaiter) Count() (waitingThreads int)

func (p *PeriodWaiter) HoldWaiters()

func (p *PeriodWaiter) IsHold() (isHold bool)

func (p *PeriodWaiter) ReleaseWaiters()

func (p *PeriodWaiter) Wait()

type Periodically ¶ added in v0.4.16

type Periodically struct {
	// contains filtered or unexported fields
}

func NewPeriodically(fn func(t time.Time), ctx context.Context, period ...time.Duration) (periodically *Periodically)

func (p *Periodically) Wait()

type PrintfFunc ¶ added in v0.4.29

type PrintfFunc func(format string, a ...any)

PrintfFunc is the signature for a printf-style function. This signature is implemented by:

• parl.Sprintf
• parl.Out parl.Outw parl.Log parl.Logw parl.Console parl.Consolew parl.Info parl.Debug parl.D parl.NoPrint
• plog.(*LogInstance) similar methods
• perrors.Errorf perrors.ErrorfPF
• fmt.Printf fmt.Sprintf fmt.Errorf
• pterm.(*StatusTerminal).Log pterm.(*StatusTerminal).LogTimeStamp

and compatible functions

type PriorityQueue ¶ added in v0.4.30

type PriorityQueue[V any, P constraints.Ordered] interface {
	// AddOrUpdate adds a new value to the prioirty queue or updates the priority of a value
	// that has changed.
	AddOrUpdate(value *V)
	// List returns the first n or default all values by priority
	List(n ...int) (valueQueue []*V)
}

PriorityQueue is a pointer-identity-to-value map of updatable values traversable by rank.

• PriorityQueue operates directly on value by caching priority from the pritority function.
• the AddOrUpdate method reprioritizes an updated value element
• V is a value reference composite type that is comparable, ie. not slice map function. Preferrably, V is interface or pointer to struct type.
• P is an ordered type such as Integer Floating-Point or string, used to rank the V values
• values are added or updated using AddOrUpdate method distinguished by (computer science) identity
• if the same comparable value V is added again, that value is re-ranked
• priority P is computed from a value V using the priorityFunc function. The piority function may be examining field values of a struct
• values can have the same rank. If they do, equal rank is provided in insertion order
• pqs.NewPriorityQueue[V any, P constraints.Ordered]
• pqs.NewRankingThreadSafe[V comparable, R constraints.Ordered]( ranker func(value *V) (rank R)))

type PrivateKey ¶ added in v0.4.26

type PrivateKey interface {
	crypto.Signer                                  // Public() Sign()
	DER() (privateKeyDer PrivateKeyDer, err error) // untyped key material, both private and public keys
	DERe() (privateKeyDer PrivateKeyDer)
	PEM() (pemBytes PemBytes, err error)
	PEMe() (pemBytes PemBytes)
	PublicKey() (publicKey PublicKey)
	Algo() (algo x509.PublicKeyAlgorithm)
	// validate ensures the private key is present, modeled after rsa.Validate
	Validate() (err error)
}

PrivateKey implements crypto.Signer and can therefore be used as tls.Certificate.PrivateKey

type PrivateKeyDer ¶ added in v0.4.26

type PrivateKeyDer []byte

PublicKeyDer is a binary encoding of a private and public key-pair

type PrivateKeyFactory ¶ added in v0.4.26

type PrivateKeyFactory interface {
	NewPrivateKey(algo x509.PublicKeyAlgorithm) (privateKey PrivateKey, err error)
}

type PublicKey ¶ added in v0.4.26

type PublicKey interface {
	DER() (publicKeyDer PublicKeyDer, err error)
	DERe() (publicKeyDer PublicKeyDer)
	PEM() (pemBytes PemBytes, err error)
	PEMe() (pemBytes PemBytes)
	Equal(x crypto.PublicKey) (isEqual bool)
	Algo() (algo x509.PublicKeyAlgorithm)
}

PublicKey contains a public key extracted from a KeyPair

type PublicKeyDer ¶ added in v0.4.26

type PublicKeyDer []byte

PublicKeyDer is a binary encoding of a public key

type Rate ¶ added in v0.4.29

type Rate interface {
	Clone() (rate Rate)
	Delta() (delta uint64)
	Duration() (duration time.Duration)
	HasValue() (hasValue bool)
	fmt.Stringer
}

Rate describes a rate datapoint.

• may be positive or negative

type RateCounterValues ¶ added in v0.4.29

type RateCounterValues interface {
	CounterValues // Get() GetReset() Value() Running() Max()
	Rates() (rates map[RateType]float64)
}

CounterValues is the consumer interface for a rate counter. The rate counter provides rates over a provided time period in a map of int64 values.

• ValueRate current rate of increase in value
• ValueMaxRate the maxmimum seen rate of increase in value
• ValueRateAverage the average rate of increase in value taken over up to 10 periods
• RunningRate the current rate of change in running, may be negative
• RunningMaxRate the max positive rate of increase seen in running
• RunningMaxDecRate the max rate of decrease in running, a 0 or negative value
• RunningAverage the average of running taken over up to 10 periods

type RateType ¶ added in v0.4.29

type RateType int

const (
	ValueRate        RateType = iota // current rate of increase in value
	ValueMaxRate                     // max seen rate of increase in value
	ValueRateAverage                 // average rate of increase in value during last 10 periods
	RunningRate
	RunningMaxRate
	RunningMaxDecRate
	// accumulated change in running over several intervals
	//	- running value goes up and down
	RunningAverage
	NotAValue // NotAValue is an internal stand-in value indicating a value not in use
)

type ServerFactory ¶ added in v0.3.0

type ServerFactory interface {
	// Adb connects to an adb adb Android Debug Bridge server on a specified tcp socket
	Adb(address AdbSocketAddress, ctx context.Context) (server Serverette)
	// AdbLocalhost connects to an adb Android Debug Bridge server on the local computer
	AdbLocalhost(ctx context.Context) (server Serverette)
}

ServerFactory describes how AdbServer objects are obtained. Such servers may use duifferent protocol implementations from Adbette

type Serverette ¶ added in v0.3.0

type Serverette interface {
	AdbAdressProvider // AdbSocketAddress()
	// DeviceSerialList lists serials for the currently online Android devices
	DeviceSerialList() (serials []AndroidSerial, err error)
	// DeviceStati lists all serials currently known to the server along with
	// their current status.
	// The two slices correspond and are of the same length
	DeviceStati() (serials []AndroidSerial, stati []AndroidStatus, err error)
	// TrackDevices emits serial numbers for devices that come online.
	// serials are sent on the serials channel.
	// if err is non-nil, set-up of failed.
	// The errs channel closes when watching stops
	// Watching is stopped by calling cancel function or when the server’s context terminates
	TrackDevices() (tracker Trackerette, err error)
}

Serverette is a generic representation of an adb server running on a host.

command-line adb: As of Android 12, Android Debug Bridge version 1.0.41 Version 32.0.0-8006631 has the following commands are supported: devices connect disconnect pair forward ppp reverse mdns push pull sync shell emu install install-multiple install-multiple-package uninstall bugreport jdwp logcat disable-verity enable-verity keygen wait-for* get-state get-serialno get-devpath remount reboot sideload root unroot usb tcpip start-server kill-server reconnect attach detach

type ServeretteFactory ¶ added in v0.3.0

type ServeretteFactory interface {
	// Adb connects to an adb adb Android Debug Bridge server on a specified tcp socket.
	// address is a string default "localhost:5037" and default port ":5037".
	// adbetter is a factory for Adbette connections.
	NewServerette(address AdbSocketAddress, adbetter Adbetter, ctx context.Context) (server Serverette)
}

ServeretteFactory is a Server connection factory for Adbette implementations

type SlowDetector ¶ added in v0.4.41

type SlowDetector struct {
	// contains filtered or unexported fields
}

SlowDetector measures latency via Start-Stop invocations and prints max latency values above threshold to stderr. Thread-safe

func NewSlowDetector(label string, slowTyp slowType, printf PrintfFunc, goGen GoGen, threshold ...time.Duration) (slowDetector *SlowDetector)

func (sd *SlowDetector) IsValid() (isValid bool)

func (sd *SlowDetector) Start(label string, value ...time.Time) (slowInvocation SlowInvocation)

func (sd *SlowDetector) Start0() (slowInvocation SlowInvocation)

func (sd *SlowDetector) Status() (s string)

func (sd *SlowDetector) Status0() (s string)

func (sd *SlowDetector) Values() (last, average, max time.Duration, hasValue bool)

type SlowDetectorCore ¶ added in v0.4.44

type SlowDetectorCore struct {
	ID slowID
	// contains filtered or unexported fields
}

SlowDetectorCore measures latency via Start-Stop invocations

• Thread-Safe and multi-threaded, parallel invocations
• Separate thread measures time of non-returning, hung invocations

func NewSlowDetectorCore(callback CbSlowDetector, slowTyp slowType, goGen GoGen, nonReturnPeriod ...time.Duration) (slowDetector *SlowDetectorCore)

func (sd *SlowDetectorCore) Start(invoLabel string, value ...time.Time) (invocation *SlowDetectorInvocation)

func (sd *SlowDetectorCore) Values() (
	last, average, max time.Duration,
	hasValue bool,
)

type SlowDetectorInvocation ¶ added in v0.4.44

type SlowDetectorInvocation struct {
	// contains filtered or unexported fields
}

SlowDetectorInvocation is a container used by SlowDetectorCore

func (sdi *SlowDetectorInvocation) Interval(label string, t ...time.Time)

func (sdi *SlowDetectorInvocation) Intervals() (intervalStr string)

func (sdi *SlowDetectorInvocation) Label() (label string)

func (sdi *SlowDetectorInvocation) Stop(value ...time.Time)

func (sdi *SlowDetectorInvocation) T0() (t0 time.Time)

func (sdi *SlowDetectorInvocation) ThreadID() (threadID ThreadID)

func (sdi *SlowDetectorInvocation) Time(t time.Time) (previousT time.Time)

type SlowDetectorThread ¶ added in v0.4.44

type SlowDetectorThread struct {
	// contains filtered or unexported fields
}

func NewSlowDetectorThread(slowTyp slowType, nonReturnPeriod time.Duration, goGen GoGen) (sdt *SlowDetectorThread)

func (sdt *SlowDetectorThread) Start(sdi *SlowDetectorInvocation)

func (sdt *SlowDetectorThread) Stop(sdi *SlowDetectorInvocation)

type SlowInvocation ¶ added in v0.4.44

type SlowInvocation interface {
	Stop(value ...time.Time)
	Interval(label string, t ...time.Time)
}

type Stack ¶ added in v0.4.20

type Stack interface {
	// thread ID 1… for the thread requesting the stack trace
	//	- ThreadID is comparable and has IsValid and String methods
	//	- ThreadID is typically an incremented 64-bit integer with
	//		main thread having ID 1
	ID() (threadID ThreadID)
	// a word indicating thread status, typically word “running”
	Status() (threadStatus ThreadStatus)
	// true if the thread is the main thread
	//	- false for a launched goroutine
	IsMain() (isMain bool)
	// A list of code locations for this thread
	//	- index [0] is the most recent code location, typically the invoker requesting the stack trace
	//	- includes invocation argument values
	Frames() (frames []pruntime.Frame)
	// the goroutine function used to launch this thread
	//	- if IsMain is true, zero-value. Check using GoFunction().IsSet()
	//	- never nil
	GoFunction() (goFunction *pruntime.CodeLocation)
	// the code location of the go statement creating this thread
	//	- if IsMain is true, zero-value. Check with Creator().IsSet()
	//	- never nil
	//	- goRoutineRef: “in goroutine 9”
	Creator() (creator *pruntime.CodeLocation, creatorID ThreadID, goRoutineRef string)
	// Shorts lists short code locations for all stack frames, most recent first:
	// Shorts("prepend") →
	//  prepend Thread ID: 1
	//  prepend main.someFunction()-pruntime.go:84
	//  prepend main.main()-pruntime.go:52
	Shorts(prepend string) (s string)
	// String is a multi-line stack trace, most recent code location first:
	//  ID: 18 IsMain: false status: running␤
	//  main.someFunction({0x100dd2616, 0x19})␤
	//  ␠␠pruntime.go:64␤
	//  cre: main.main-pruntime.go:53␤
	fmt.Stringer
}

Stack contains a stack trace parsed into basic type only datapoints

• stack trace from [pdebug.Stack]

type Statuser ¶ added in v0.3.0

type Statuser interface {
	Set(status string) (statuser Statuser)
	Shutdown()
}

Statuser prints threads that do not react within a specified time frame. This is useful during early multi-threaded design when it is uncertain why work is not progressing. Is it your code or their code? Once the program concludes without hung threads, Tracer is the better tool to identify issues.

type StatuserFactory ¶ added in v0.3.0

type StatuserFactory interface {
	NewStatuser(useStatuser bool, d time.Duration) (statuser Statuser)
}

type SubGo ¶ added in v0.4.15

type SubGo interface {
	// Go returns a [Go] object managing a thread of the GoGroup thread-group
	// by providing the g value as a go-statement function argument.
	Go() (g Go)
	// SubGo returns a thread-group whose fatal errors go to Go’s parent.
	//   - both non-fatal and fatal errors in SubGo threads are sent to Go’s parent
	// 		like Go.AddError and Go.Done.
	//		- therefore, when a SubGo thread fails, the application will typically exit.
	//		- by awaiting SubGo, Go can delay its exit until SubGo has terminated
	//   - the SubGo thread-group terminates when the its thread exits
	SubGo(onFirstFatal ...GoFatalCallback) (subGo SubGo)
	// SubGroup creates a sub-ordinate GoGroup.
	//	- SubGroup fatal and non-fatal errors are sent to the parent GoGroup.
	//	-	SubGroup-context initiated Cancel only affect threads in the SubGroup thread-group
	//	- parent-initiated Cancel terminates SubGroup threads
	//	- SubGroup only awaits SubGroup threads
	//	- parent await also awaits SubGroup threads
	SubGroup(onFirstFatal ...GoFatalCallback) (subGroup SubGroup)
	// Wait waits for all threads of this thread-group to terminate.
	Wait()
	// returns a channel that closes on subGo end similar to Wait
	WaitCh() (ch AwaitableCh)
	// EnableTermination controls temporarily preventing the GoGroup from
	// terminating.
	// EnableTermination is initially true.
	//	- invoked with no argument returns the current state of EnableTermination
	//	- invoked with [AllowTermination] again allows for termination and
	//		immediately terminates the thread-group if no threads are currently running.
	//	- invoked with [PreventTermination] allows for the number of managed
	//		threads to be temporarily zero without terminating the thread-group
	EnableTermination(allowTermination ...bool) (mayTerminate bool)
	// Cancel terminates the threads in this and subordinate thread-groups.
	Cancel()
	// Context will Cancel when the parent context Cancels.
	// Subordinate thread-groups do not Cancel this context.
	Context() (ctx context.Context)
	// the available data for all threads
	Threads() (threads []ThreadData)
	// threads that have been named ordered by name
	NamedThreads() (threads []ThreadData)
	// SetDebug enables debug logging on this particular instance
	//   - parl.NoDebug
	//   - parl.DebugPrint
	//   - parl.AggregateThread
	SetDebug(debug GoDebug, log ...PrintfFunc)
	fmt.Stringer
}

type SubGroup ¶ added in v0.4.29

type SubGroup interface {
	SubGo
	// Ch returns a receive channel for fatal errors if this SubGo has LocalChannel option.
	Ch() (ch <-chan GoError)
	// FirstFatal allows to await or inspect the first thread terminating with error.
	// it is valid if this SubGo has LocalSubGo or LocalChannel options.
	// To wait for first fatal error using multiple-semaphore mechanic:
	//  firstFatal := g0.FirstFatal()
	//  for {
	//    select {
	//    case <-firstFatal.Ch():
	//    …
	// To inspect first fatal:
	//  if firstFatal.DidOccur() …
	FirstFatal() (firstFatal *OnceWaiterRO)
}

type SyncAdd ¶ added in v0.4.20

type SyncAdd interface {
	Add(delta int)
}

SyncWait provides sync.WaitGroup.Add()

type SyncDone ¶ added in v0.4.20

type SyncDone interface {
	Done()
}

SyncDone provides sync.WaitGroup.Done()

type SyncWait ¶ added in v0.4.20

type SyncWait interface {
	Wait()
}

SyncWait provides sync.WaitGroup.Wait()

type TFunc ¶ added in v0.4.60

type TFunc[T any] func() (value T, err error)

TFunc is a function that returns value, err and may panic

type TResult ¶ added in v0.4.60

type TResult[T any] struct {
	Value   T
	IsPanic bool
	Err     error
}

TResult is a value-container for value, isPanic and error

func NewTResult[T any](tFunc ...TFunc[T]) (tResult *TResult[T])

func NewTResult3[T any](value *T, isPanic *bool, errp *error) (tResult *TResult[T])

type ThreadData ¶ added in v0.4.43

type ThreadData interface {
	// threadID is the ID of the running thread assigned by the go runtime
	//	- IsValid method checks if value is present
	//	- zero value is empty string
	//	- .ThreadID().String(): "5"
	ThreadID() (threadID ThreadID)
	// createLocation is the code line of the go statement function-call
	// creating the goroutine thread
	// - IsSet method checks if value is present
	//	- Create().Short(): "g0.(*SomeType).SomeCode()-thread-data_test.go:73"
	Create() (createLocation *pruntime.CodeLocation)
	// Func returns the code line of the function of the running thread.
	// - IsSet method checks if value is present
	//	- .Func().Short(): "g0.(*SomeType).SomeFunction()-thread-data_test.go:80"
	Func() (funcLocation *pruntime.CodeLocation)
	// optional thread-name assigned by consumer
	//	- zero-value: empty string "" for threads that have not been named
	Name() (label string)
	// Short returns a short description of the thread "label:threadID" or fmt.Stringer
	//	- "myThreadName:4"
	//	- zero-value: "[empty]" ThreadDataEmpty
	//	- nil value: "threadData:nil" ThreadDataNil
	Short() (short string)
	// all non-empty fields: [label]:[threadID]_func:[Func]_cre:[Create]
	//	- "myThreadName:5_func:g0.(*SomeType).SomeFunction()-thread-data_test.go:80_cre:g0.(*SomeType).SomeCode()-thread-data_test.go:73"
	//	- zero-value: "[empty]" ThreadDataEmpty
	fmt.Stringer
}

ThreadData is information about a Go object thread.

• initially, only Create is present
• Name is only present for threads that have been named

type ThreadID ¶ added in v0.4.9

type ThreadID uint64

ThreadID is an opaque type that uniquley identifies a thread, ie. a goroutine.

• goid.GoID obtains ThreadID for the executing thread.
• in runtime.g, goid is uint64
• ThreadID is comparable, ie. can be used as a map key.
• ThreadID is fmt.Stringer
• ThreadID has IsValid method

func (threadID ThreadID) IsValid() (isValid bool)

func (threadID ThreadID) String() (s string)

type ThreadStatus ¶ added in v0.4.12

type ThreadStatus string

ThreadStatus indicates the current stat of a thread most often it is "running"

type Timer ¶

type Timer struct {
	Label string
	// contains filtered or unexported fields
}

Timer is a simple request timer

func NewTimer(label string) (t *Timer)

func (t *Timer) End() (d time.Duration)

func (t *Timer) Endms() (ms string)

type Tracer ¶ added in v0.3.0

type Tracer interface {
	// AssignTaskToThread assigns a Thread to a task
	AssignTaskToThread(threadID ThreadID, task TracerTaskID) (tracer Tracer)
	// RecordTaskEvent adds an event to the task threadID is currently assigned to.
	// If threadID is not assigned, a new task is created.
	RecordTaskEvent(threadID ThreadID, text string) (tracer Tracer)
	// Records returns the current map of tasks and their events
	Records(clear bool) (records map[TracerTaskID][]TracerRecord)
}

Tracer lists events in terms of tasks rather than per time or thread. A task is executed by threads assigned to it. Threads are uniquely identified by threadID. A task can have zero or one threads assigned at any one time. A thread can be assigned to zero or one tasks. Each task has an ID, a name and a list of events and thread assignments Tracer can record branching in the code and return that for a particular item being processed. For an item processed incorrectly, or when threads hang, Tracer will find unfavorable branching and last known locations.

type TracerFactory ¶ added in v0.3.0

type TracerFactory interface {
	// NewTracer creates tracer storage.
	// use false indicates a nil Tracer whose output will not be used
	NewTracer(use bool) (tracer Tracer)
}

type TracerRecord ¶ added in v0.3.0

type TracerRecord interface {
	Values() (at time.Time, text string)
}

type TracerTaskID ¶ added in v0.3.0

type TracerTaskID string

type Trackerette ¶ added in v0.3.0

type Trackerette interface {
	// Serials emit serial number as online devices become available
	Serials() (serials <-chan AndroidSerial)
	// Errs is available once Serials close. It returns any errors
	Errs() (err error)
	// Cancel shuts down the Tracker
	Cancel()
}

Trackerette represents a server connection emitting device serial numbers

type UniqueID ¶ added in v0.4.26

type UniqueID[T ~string] uint64

UniqueID is an executable-invocation-unique identifier generator. The format is a named-type small-integer numeric string suitable to distinguish multiple instances of a type. The type is ordered and can be converted to string.

Usage:

type MyType string
var generator parl.UniqueID[MyType]
someID := generator.ID()

func (u *UniqueID[T]) ID() (unique T)

type UniqueIDTypedUint64 ¶ added in v0.4.29

type UniqueIDTypedUint64[T ~uint64] uint64

UniqueIDTypedUint64 generates integer-based opaque uniquely-typed IDs. Thread-safe.

• Different named-type series have their own unique type and can be told apart.
• The named types are ordered integer-based with String method implementing fmt.Stringer.

Usage:

type T uint64
var generator parl.UniqueIDTypedUint64[T]
func (t T) String() string { return generator.StringT(t) }
someID := generator.ID()

var GoEntityIDs UniqueIDTypedUint64[GoEntityID]

func (u *UniqueIDTypedUint64[T]) ID() (uniqueT T)

func (u *UniqueIDTypedUint64[T]) StringT(t T) (s string)

type UniqueIDUint64 ¶ added in v0.4.29

type UniqueIDUint64 uint64

UniqueIDUint64 generates executable-invocation-unique uint64 numbers 1… Different series of uint64 from different generators does not have identity, ie. they cannot be told apart. Consider UniqueIDTypedUint64 to have identity.

Usage:

var generator parl.UniqueIDUint64
id := generator.ID()

func (u *UniqueIDUint64) ID() (uniqueUint64 uint64)

func (u *UniqueIDUint64) String() (s string)

type UniqueIDint ¶ added in v0.4.30

type UniqueIDint int

UniqueIDUint64 generates executable-invocation-unique uint64 numbers 1… Different series of uint64 from different generators does not have identity, ie. they cannot be told apart. Consider UniqueIDTypedUint64 to have identity.

Usage:

var generator parl.UniqueIDUint64
id := generator.ID()

func (u *UniqueIDint) ID() (uniqueID int)

func (u *UniqueIDint) String() (s string)

type WaitGroup ¶ added in v0.3.0

type WaitGroup struct {
	sync.WaitGroup // Wait()
	// contains filtered or unexported fields
}

parl.WaitGroup is like a sync.Waitgroup that can be inspected. The Waiting method returns the number of threads waited for. parl.WaitGroup requires no initialization.

var wg parl.WaitGroup
wg.Add(1)
…
wg.Waiting()

func NewWaitGroup() (waitGroup *WaitGroup)

func (wg *WaitGroup) Add(delta int)

func (wg *WaitGroup) Count() (remaining int)

func (wg *WaitGroup) Counters() (adds int, dones int)

func (wg *WaitGroup) Done()

func (wg *WaitGroup) DoneBool() (isExit bool)

func (wg *WaitGroup) IsZero() (isZero bool)

func (wg *WaitGroup) String() (s string)

type WaitGroupCh ¶ added in v0.4.148

type WaitGroupCh struct {
	// contains filtered or unexported fields
}

WaitGroupCh is like a sync.WaitGroup with channel-wait mechanic. therefore, unlike sync, WaitGroupCh is wait-free and observable. WaitGroupCh waits for a collection of goroutines to finish. The main goroutine increments the counter for each goroutine. Then each of the goroutines decrements the counter until zero. Wait or Ch can be used to block until all goroutines have finished. A WaitGroup must not be copied after first use. In the terminology of the Go memory model, decrementing “synchronizes before” the return of any Wait call or Ch read that it unblocks.

• counter is increased by WaitGroupCh.Add or WaitGroupCh.Count
• counter is decreased by WaitGroupCh.Done WaitGroupCh.Add WaitGroupCh.DoneBool or WaitGroupCh.Count
• counter zero is awaited by WaitGroupCh.Ch or WaitGroupCh.Wait
• observability if provided by WaitGroupCh.Count WaitGroupCh.DoneBool and WaitGroupCh.IsZero
• panic: negative counter from invoking decreasing methods is panic
• panic: adjusting away from zero after invocation of WaitGroupCh.Ch or WaitGroupCh.Wait is panic. NOTE: all Add should happen prior to invoking Ch or Wait
• —
• WaitGroupCh is wait-free, observable, initialization-free, thread-safe
• channel-wait mechanic allows the consumer to be wait-free Progress by the consumer-thread is not prevented since:
• — the channel can be read non-blocking
• — consumers can wait for multiple channel events
• — consumers are not contending for a lock with any other thread
• WaitGroupCh method-set is a superset of sync.WaitGroup
• there are race conditions:
• — writing a zero-counter record with unclosed channel, therefore closing the channel
• — reading channel event while a zero-counter record exists, therefore closing the channel
• — impact is that a panic might be missed

Usage:

var w parl.WaitGroupCh
w.Add(1)
go someFunc(&w)
…
<-w.Ch()
func someFunc(w parl.Doneable) {
  defer w.Done()

func (w *WaitGroupCh) Add(delta int)

func (w *WaitGroupCh) Ch() (awaitableCh AwaitableCh)

func (w *WaitGroupCh) Count(delta ...int) (currentCount, totalAdds int)

func (w *WaitGroupCh) Done()

func (w *WaitGroupCh) DoneBool() (isExit bool)

func (w *WaitGroupCh) IsZero() (isZero bool)

func (w *WaitGroupCh) Reset() (w2 *WaitGroupCh)

func (w *WaitGroupCh) String() (s string)

func (w *WaitGroupCh) Wait()

type Waitable ¶ added in v0.4.12

type Waitable interface {
	Wait() // similar to sync.WaitGroup.Wait()
}

Waitable is the invoker’s Wait part of sync.WaitGroup and other implementations. Waitable is a many-to-many relation. Waitable allows the caller to await exit and free of all invocations.

waitsForLots parl.WaitGroup
shutsDownLots parl.OnceChan
… = NewSomething(&waitsForLots, &shutsDownLots)
go someThread(&waitsForLots, &shutsDownLots)
func someThread(Doneable w, context.Context ctx) {
  defer w.Done()
  w.Add(2)
  go somethingElse()

type WaitedOn ¶ added in v0.4.20

type WaitedOn interface {
	SyncAdd
	SyncDone
	DoneBool() (isExit bool)
	IsZero() (isZero bool)
}

type Waiter ¶ added in v0.4.20