Coverage for src / lexigram / contracts / core / concurrency_protocols.py: 0%

1"""Concurrency protocol class definitions.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, Protocol, runtime_checkable 

6 

7from lexigram.contracts.core.concurrency_enums import ExecutionStrategy 

8 

9 

10@runtime_checkable 

11class TaskManagerProtocol(Protocol): 

12 """Protocol for task management with priority and tagging support.""" 

13 

14 def create_critical_task( 

15 self, 

16 coro: Any, 

17 *, 

18 name: str | None = None, 

19 ) -> Any: # asyncio.Task[Any] in real implementation 

20 """Create a critical task that should complete before shutdown.""" 

21 ... 

22 

23 def create_background_task( 

24 self, 

25 coro: Any, # Awaitable[Any] in real implementation 

26 *, 

27 name: str | None = None, 

28 ) -> Any: # asyncio.Task[Any] in real implementation 

29 """Create a background task that can be cancelled on shutdown.""" 

30 ... 

31 

32 async def shutdown_gracefully( 

33 self, 

34 critical_timeout: float = 10.0, 

35 background_timeout: float = 2.0, 

36 ) -> None: 

37 """Shutdown tasks gracefully with different timeouts.""" 

38 ... 

39 

40 def get_task_counts(self) -> dict[str, int]: 

41 """Get current task counts for monitoring.""" 

42 ... 

43 

44 

45@runtime_checkable 

46class DispatcherProtocol(Protocol): 

47 """Protocol for dispatching sync and async functions with automatic detection. 

48 

49 Sync functions are offloaded to a thread pool with context propagation. 

50 Async functions run directly on the event loop. 

51 """ 

52 

53 async def run( 

54 self, 

55 func: Any, 

56 *args: Any, 

57 executor: Any | None = None, 

58 timeout: float | None = None, 

59 **kwargs: Any, 

60 ) -> Any: 

61 """Execute a function, automatically choosing sync or async path. 

62 

63 Args: 

64 func: Sync or async callable to execute. 

65 args: Positional arguments forwarded to func. 

66 executor: Optional executor override for sync functions. 

67 timeout: Optional per-call timeout in seconds. 

68 kwargs: Keyword arguments forwarded to func. 

69 

70 Returns: 

71 The return value of func. 

72 

73 Raises: 

74 RuntimeError: If the dispatcher is draining. 

75 """ 

76 ... 

77 

78 async def run_many( 

79 self, 

80 tasks: list[tuple[Any, tuple[Any, ...], dict[str, Any]]], 

81 concurrency: int | None = None, 

82 timeout: float | None = None, 

83 ) -> list[Any]: 

84 """Execute multiple functions concurrently. 

85 

86 Args: 

87 tasks: List of (func, args, kwargs) tuples. 

88 concurrency: Max concurrent executions. None means unlimited. 

89 timeout: Per-task timeout in seconds. 

90 

91 Returns: 

92 Results in the same order as tasks. 

93 """ 

94 ... 

95 

96 async def shutdown( 

97 self, 

98 wait: bool = True, 

99 drain: bool = False, 

100 drain_timeout: float | None = None, 

101 ) -> None: 

102 """Shutdown the dispatcher and release thread pools. 

103 

104 Args: 

105 wait: Wait for running tasks to complete. 

106 drain: Stop accepting new tasks before shutdown. 

107 drain_timeout: Max seconds to wait for active tasks to drain. 

108 """ 

109 ... 

110 

111 @property 

112 def is_draining(self) -> bool: 

113 """True if the dispatcher is rejecting new tasks.""" 

114 ... 

115 

116 @property 

117 def active_task_count(self) -> int: 

118 """Number of currently executing tasks.""" 

119 ... 

120 

121 

122@runtime_checkable 

123class ParallelProtocol(Protocol): 

124 """Protocol for structured parallelism with safe error handling.""" 

125 

126 @classmethod 

127 async def execute( 

128 cls, 

129 *tasks: Any, 

130 strategy: ExecutionStrategy = ExecutionStrategy.GATHER, 

131 ) -> list[Any]: 

132 """Execute tasks in parallel with the specified strategy.""" 

133 ... 

134 

135 

136@runtime_checkable 

137class ChannelProtocol(Protocol): 

138 """Protocol for typed async channels providing backpressure-aware communication. 

139 

140 A channel decouples producers and consumers with an optional bounded 

141 buffer. Closing a channel signals that no more items will be sent. 

142 """ 

143 

144 async def send(self, item: Any) -> None: 

145 """Send an item into the channel. 

146 

147 Blocks if the channel's buffer is full until space is available. 

148 

149 Args: 

150 item: Item to enqueue. 

151 

152 Raises: 

153 ChannelClosedError: If the channel has already been closed. 

154 ChannelFullError: If the channel is bounded and at capacity 

155 and the caller does not wish to wait. 

156 """ 

157 ... 

158 

159 async def receive(self) -> Any: 

160 """Receive the next item from the channel. 

161 

162 Blocks until an item is available or the channel is closed. 

163 

164 Returns: 

165 The next item from the channel. 

166 

167 Raises: 

168 ChannelClosedError: If the channel is closed and empty. 

169 """ 

170 ... 

171 

172 async def close(self) -> None: 

173 """Close the channel. 

174 

175 After closing, no further items may be sent. Pending items can 

176 still be received until the buffer is drained. 

177 """ 

178 ... 

179 

180 @property 

181 def closed(self) -> bool: 

182 """True if the channel has been closed.""" 

183 ... 

184 

185 

186__all__ = [ 

187 "ChannelProtocol", 

188 "DispatcherProtocol", 

189 "ParallelProtocol", 

190 "TaskManagerProtocol", 

191]