Buffer refactor

.github/workflows/tests-unit.yml

@@ -15,7 +15,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
         os:
           - ubuntu-latest
           - macos-latest

examples/managers/blender/blender.py

@@ -16,7 +16,7 @@
     PaymentPlatformNegotiator,
     PoolActivityManager,
     RefreshingDemandManager,
-    ScoringBuffer,
+    ScoringBufferPlugin,
     WorkContext,
     WorkResult,
     retry,
@@ -58,7 +58,7 @@ async def run_on_golem(
         demand_manager.get_initial_proposal,
         plugins=[
             NegotiatingPlugin(proposal_negotiators=negotiators),
-            ScoringBuffer(
+            ScoringBufferPlugin(
                 min_size=3, max_size=5, fill_concurrency_size=3, proposal_scorers=scorers
             ),
         ],

examples/managers/flexible_negotiation.py

@@ -3,7 +3,7 @@
 
 from golem.managers import (
     BlacklistProviderIdPlugin,
-    Buffer,
+    BufferPlugin,
     DefaultAgreementManager,
     DefaultProposalManager,
     NegotiatingPlugin,
@@ -48,14 +48,14 @@ async def main():
         golem,
         demand_manager.get_initial_proposal,
         plugins=[
-            Buffer(
+            BufferPlugin(
                 min_size=10,
                 max_size=1000,
                 fill_concurrency_size=5,
             ),
             BlacklistProviderIdPlugin(BLACKLISTED_PROVIDERS),
             NegotiatingPlugin(proposal_negotiators=[PaymentPlatformNegotiator()]),
-            Buffer(
+            BufferPlugin(
                 min_size=3,
                 max_size=5,
                 fill_concurrency_size=3,

examples/managers/mid_agreement_payments.py

@@ -3,7 +3,7 @@
 from datetime import datetime, timedelta
 
 from golem.managers import (
-    Buffer,
+    BufferPlugin,
     DefaultAgreementManager,
     DefaultProposalManager,
     MidAgreementPaymentsNegotiator,
@@ -91,7 +91,7 @@ async def main():
                     ),
                 ]
             ),
-            Buffer(
+            BufferPlugin(
                 min_size=1,
                 max_size=4,
                 fill_concurrency_size=2,

examples/managers/proposal_plugins.py

@@ -5,7 +5,7 @@
 
 from golem.managers import (
     BlacklistProviderIdPlugin,
-    Buffer,
+    BufferPlugin,
     DefaultAgreementManager,
     DefaultProposalManager,
     LinearAverageCostPricing,
@@ -19,7 +19,7 @@
     RefreshingDemandManager,
     RejectIfCostsExceeds,
     RejectProposal,
-    ScoringBuffer,
+    ScoringBufferPlugin,
     SequentialWorkManager,
     WorkContext,
     WorkResult,
@@ -78,7 +78,7 @@ async def main():
         golem,
         demand_manager.get_initial_proposal,
         plugins=[
-            Buffer(
+            BufferPlugin(
                 min_size=10,
                 max_size=1000,
             ),
@@ -101,7 +101,7 @@ async def main():
                     else None,
                 )
             ),
-            ScoringBuffer(
+            ScoringBufferPlugin(
                 min_size=3,
                 max_size=5,
                 fill_concurrency_size=3,

examples/task_api_draft/examples/yacat.py

@@ -1,7 +1,6 @@
 import asyncio
 from collections import defaultdict
 from typing import (
-    Any,
     AsyncIterator,
     Callable,
     DefaultDict,
@@ -221,9 +220,7 @@ async def main() -> None:
 
 #############################################
 #   NOT REALLY INTERESTING PARTS OF THE LOGIC
-async def close_agreement_repeat_task(
-    func: Callable, args: Tuple[Activity, Any], e: Exception
-) -> None:
+async def close_agreement_repeat_task(func: Callable, args: Tuple, e: Exception) -> None:
     activity, task = args
     tasks_queue.put_nowait(task)
     print("Task failed on", activity)

examples/task_api_draft/task_api/activity_pool.py

@@ -4,6 +4,7 @@
 
 from golem.pipeline import InputStreamExhausted
 from golem.resources import Activity
+from golem.utils.asyncio import cancel_and_await
 
 
 class ActivityPool:
@@ -94,7 +95,7 @@ async def _activity_destroyed_cleanup(
             return
 
         await activity.wait_destroyed()
-        manager_task.cancel()
+        await cancel_and_await(manager_task)
 
     async def _get_next_idle_activity(
         self, activity_stream: AsyncIterator[Union[Activity, Awaitable[Activity]]]

examples/task_api_draft/task_api/execute_tasks.py

@@ -31,10 +31,10 @@ async def random_score(proposal: Proposal) -> float:
 
 def close_agreement_repeat_task(
     task_stream: TaskDataStream[TaskData],
-) -> Callable[[Callable, Tuple[Activity, TaskData], Exception], Awaitable[None]]:
+) -> Callable[[Callable, Tuple, Exception], Awaitable[None]]:
     async def on_exception(
-        func: Callable[[Activity, TaskData], Awaitable[TaskResult]],
-        args: Tuple[Activity, TaskData],
+        func: Callable,
+        args: Tuple,
         e: Exception,
     ) -> None:
         activity, in_data = args

golem/event_bus/in_memory/event_bus.py

@@ -5,7 +5,7 @@
 from typing import Awaitable, Callable, DefaultDict, List, Optional, Tuple, Type
 
 from golem.event_bus.base import Event, EventBus, EventBusError, TEvent
-from golem.utils.asyncio import create_task_with_logging
+from golem.utils.asyncio import cancel_and_await, create_task_with_logging
 from golem.utils.logging import get_trace_id_name, trace_span
 
 logger = logging.getLogger(__name__)
@@ -49,7 +49,7 @@ async def stop(self):
         await self._event_queue.join()
 
         if self._process_event_queue_loop_task is not None:
-            self._process_event_queue_loop_task.cancel()
+            await cancel_and_await(self._process_event_queue_loop_task)
             self._process_event_queue_loop_task = None
 
     @trace_span(show_results=True)

golem/managers/__init__.py

@@ -23,7 +23,7 @@
 from golem.managers.payment import PayAllPaymentManager
 from golem.managers.proposal import (
     BlacklistProviderIdPlugin,
-    Buffer,
+    BufferPlugin,
     DefaultProposalManager,
     LinearAverageCostPricing,
     LinearCoeffsCost,
@@ -37,7 +37,7 @@
     ProposalScoringMixin,
     RandomScore,
     RejectIfCostsExceeds,
-    ScoringBuffer,
+    ScoringBufferPlugin,
 )
 from golem.managers.work import (
     ConcurrentWorkManager,
@@ -73,7 +73,7 @@
     "PayAllPaymentManager",
     "DefaultProposalManager",
     "BlacklistProviderIdPlugin",
-    "Buffer",
+    "BufferPlugin",
     "PaymentPlatformNegotiator",
     "MidAgreementPaymentsNegotiator",
     "NegotiatingPlugin",
@@ -86,7 +86,7 @@
     "LinearCoeffsCost",
     "PropertyValueLerpScore",
     "RandomScore",
-    "ScoringBuffer",
+    "ScoringBufferPlugin",
     "SequentialWorkManager",
     "ConcurrentWorkManager",
     "WorkManagerPluginsMixin",

golem/managers/demand/refreshing.py

@@ -10,9 +10,8 @@
 from golem.payload import defaults as payload_defaults
 from golem.resources import Allocation, Demand, Proposal
 from golem.resources.demand.demand_builder import DemandBuilder
-from golem.utils.asyncio import create_task_with_logging
+from golem.utils.asyncio import ErrorReportingQueue, create_task_with_logging
 from golem.utils.logging import get_trace_id_name, trace_span
-from golem.utils.queue import ErrorReportingQueue
 
 logger = logging.getLogger(__name__)
 

golem/managers/mixins.py

@@ -3,7 +3,7 @@
 from typing import Generic, List, Optional, Sequence
 
 from golem.managers.base import ManagerException, TPlugin
-from golem.utils.asyncio import create_task_with_logging
+from golem.utils.asyncio import cancel_and_await, create_task_with_logging
 from golem.utils.logging import get_trace_id_name, trace_span
 
 logger = logging.getLogger(__name__)
@@ -29,7 +29,7 @@ async def stop(self) -> None:
             raise ManagerException("Already stopped!")
 
         if self._background_loop_task is not None:
-            self._background_loop_task.cancel()
+            await cancel_and_await(self._background_loop_task)
             self._background_loop_task = None
 
     def is_started(self) -> bool:

golem/managers/proposal/__init__.py

@@ -1,7 +1,7 @@
 from golem.managers.proposal.default import DefaultProposalManager
 from golem.managers.proposal.plugins import (
     BlacklistProviderIdPlugin,
-    Buffer,
+    BufferPlugin,
     LinearAverageCostPricing,
     LinearCoeffsCost,
     LinearPerCpuAverageCostPricing,
@@ -14,13 +14,13 @@
     ProposalScoringMixin,
     RandomScore,
     RejectIfCostsExceeds,
-    ScoringBuffer,
+    ScoringBufferPlugin,
 )
 
 __all__ = (
     "DefaultProposalManager",
     "BlacklistProviderIdPlugin",
-    "Buffer",
+    "BufferPlugin",
     "PaymentPlatformNegotiator",
     "MidAgreementPaymentsNegotiator",
     "NegotiatingPlugin",
@@ -33,5 +33,5 @@
     "LinearPerCpuCoeffsCost",
     "PropertyValueLerpScore",
     "RandomScore",
-    "ScoringBuffer",
+    "ScoringBufferPlugin",
 )

golem/managers/proposal/plugins/__init__.py

@@ -1,5 +1,5 @@
 from golem.managers.proposal.plugins.blacklist import BlacklistProviderIdPlugin
-from golem.managers.proposal.plugins.buffer import Buffer
+from golem.managers.proposal.plugins.buffer import BufferPlugin
 from golem.managers.proposal.plugins.linear_coeffs import LinearCoeffsCost, LinearPerCpuCoeffsCost
 from golem.managers.proposal.plugins.negotiating import (
     MidAgreementPaymentsNegotiator,
@@ -14,12 +14,12 @@
     PropertyValueLerpScore,
     ProposalScoringMixin,
     RandomScore,
-    ScoringBuffer,
+    ScoringBufferPlugin,
 )
 
 __all__ = (
     "BlacklistProviderIdPlugin",
-    "Buffer",
+    "BufferPlugin",
     "PaymentPlatformNegotiator",
     "MidAgreementPaymentsNegotiator",
     "NegotiatingPlugin",
@@ -32,5 +32,5 @@
     "LinearPerCpuCoeffsCost",
     "PropertyValueLerpScore",
     "RandomScore",
-    "ScoringBuffer",
+    "ScoringBufferPlugin",
 )