diff --git a/Dockerfile b/Dockerfile index eac78b9c5..3e6855c9c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM eclipse-temurin:24-jdk +FROM eclipse-temurin:25-jdk COPY nb5/target/nb5.jar nb5.jar ENTRYPOINT ["java","--enable-preview", "-XX:+UseZGC", "-jar", "nb5.jar"] diff --git a/arch_planning.md b/arch_planning.md new file mode 100644 index 000000000..cc175b952 --- /dev/null +++ b/arch_planning.md @@ -0,0 +1,438 @@ +# NoSQLBench Architectural Planning - Consolidated Analysis + +## Status Overview (Updated 2025-09-29) + +**Current Branch**: `jshook/refactorings1` +**Overall Progress**: Phase 1 Step 1 Complete, Step 2 Starting + +| Phase | Status | Branch | Progress | +|-------|--------|--------|----------| +| **Phase 1: Core Consolidation** | 🔄 In Progress | refactorings1 | ████░░░░░░ 40% | +| └─ Step 1.1: Activity Consolidation | ✅ Complete | refactorings1 | ██████████ 100% | +| └─ Step 1.2: ActivityConfig Migration | ⚪ Not Started | refactorings1 | ░░░░░░░░░░ 0% | +| └─ Step 1.3: Lifecycle/Error Handling | ⚪ Not Started | refactorings1 | ░░░░░░░░░░ 0% | +| **Phase 2: Operation Pipeline** | ⚪ Not Started | - | ░░░░░░░░░░ 0% | +| **Phase 3: Enhancements** | ⚪ Not Started | - | ░░░░░░░░░░ 0% | +| **Phase 4: Polish** | ⚪ Not Started | - | ░░░░░░░░░░ 0% | + +**Legend**: ✅ Complete | 🔄 In Progress | ⚪ Not Started + +**Note**: The `jshook/nb523` branch serves as a reference implementation showing the intended architectural changes. This `refactorings1` branch will apply those changes incrementally in a cleaner, more systematic way. + +--- + +## Executive Summary + +This document consolidates the analysis from three perspectives of the nb523 branch refactoring: +1. **nb523-branch-analysis.md** - Design intentions and commit analysis +2. **refactorings.md** - Detailed implementation steps with code examples +3. **branch_changes.md** - Net architectural changes and impact analysis + +The refactoring represents a fundamental simplification of NoSQLBench's architecture, focusing on consolidation, standardization, and improved developer experience. + +## Common Themes Across All Documents + +### 1. Activity Class Hierarchy Consolidation +**Status**: ✅ Complete on jshook/refactorings1 branch +**Unanimous Agreement:** +- Elimination of `SimpleActivity`, `StandardActivity`, and `Activity` interface +- Creation of single unified `Activity` class +- Location: `io.nosqlbench.engine.api.activityapi.core.Activity` +- Result: ~1,300 lines net removed from codebase (767 line unified Activity class) + +**Key Benefits Identified:** +- Reduced cognitive load for developers +- Clearer inheritance structure +- Single source of truth for activity behavior +- Simplified maintenance + +### 2. Configuration System Modernization +**Status**: ⚪ To be implemented on refactorings1 (reference impl in nb523) +**Planned Changes:** +- Removal of `ActivityDef` (272 lines) +- Introduction of `ActivityConfig` extending `NBConfiguration` (113 lines) +- Implementation of `NBConfigurable` and `NBReconfigurable` interfaces +- Standardized configuration validation + +**Implementation Details:** +- Type-safe parameter access +- Built-in validation framework +- Dynamic reconfiguration support +- Backward compatibility through NBConfiguration + +**Reference**: See `jshook/nb523` branch for example implementation approach + +### 3. Operation Synthesis Pipeline Enhancement +**Status**: ⚪ Not started (planned in original nb523 analysis) +**Shared Components:** +- `OpResolution` - Central coordinator for operation synthesis (167 lines planned) +- `OpResolverBank` - Managing multiple operation resolvers (76 lines planned) +- `AdapterResolver` - Driver adapter lifecycle management (67 lines planned) +- `DispenserResolver` - Creating operation dispensers (40 lines planned) +- `ParsedOpResolver` - Template to parsed operation conversion (44 lines planned) + +**Common Goals:** +- Better separation of concerns +- Circular dependency detection +- Lazy initialization +- Improved error reporting + +### 4. Lifecycle and State Management +**Agreed Improvements:** +- Clear state transitions +- Better thread safety +- Lifecycle state validation +- Improved error handling during state changes + +### 5. Code Organization and Documentation +**Universal Changes:** +- Addition of style configurations (IntelliJ, EditorConfig) +- License header updates +- Improved inline documentation +- Visual architecture diagrams (scaffold.svg/png) + +## Themes Present in Two Documents + +### Present in nb523-branch-analysis.md and branch_changes.md: +1. **ActivityWiring Class Introduction** + - Centralizes component wiring + - Manages dispensers for motors, inputs, actions, outputs, filters + - 89 lines of focused responsibility + +2. **ActivityMetrics Separation** + - Extracted from main activity class + - 176 lines of dedicated metrics management + - Better metric categorization + +3. **FieldVerifier System** + - 260 lines of validation logic + - Type checking and constraint validation + - Enhanced error reporting + +### Present in refactorings.md and branch_changes.md: +1. **Motor System Updates** + - CoreMotor extends NBBaseComponent + - Implements NBReconfigurable + - Simplified constructor + - Better sync/async separation + +2. **Diagnostics Class Addition** + - 59 lines for diagnostic capabilities + - Activity introspection + - Enhanced debugging + +### Present in nb523-branch-analysis.md and refactorings.md: +1. **Dry-Run Capability Enhancement** + - Operation preview without execution + - Better testing and validation + - Developer-friendly debugging + +2. **Space API Alignment** + - Consistent interface across adapters + - Better lifecycle management + - Standardized configuration + +## Unique Details by Document + +### Unique to nb523-branch-analysis.md: +1. **Commit-by-Commit Analysis** + - Specific commit hashes and progression + - 30+ commits tracked + - Intermediate state documentation + +2. **Technical Debt Metrics** + - ~1,900 lines of code removed + - Observer pattern elimination details + +3. **Risk Assessment** + - Breaking change identification + - Compatibility concerns + - Performance validation needs + +### Unique to refactorings.md: +1. **Atomic Implementation Steps** + - Line-by-line code changes + - Specific file paths and line numbers + - Test commands for each step + +2. **Extensive Code Examples** + - Complete class implementations + - Before/after comparisons + - Test case examples + +3. **Implementation Tools:** + - OpCache implementation for performance + - SpaceCache manager + - LifecycleStateMachine + - ConfigurationValidator utility + - Field resolver strategies + - Operation pooling mechanisms + +4. **Rollback Procedures** + - Quick rollback commands + - Partial rollback strategies + - Phase-by-phase reversal + +5. **5-Week Implementation Timeline** + - Week-by-week breakdown + - Detailed checklists + - Success criteria + +### Unique to branch_changes.md: +1. **File Statistics** + - Exact line counts for additions/removals + - File path changes and reorganization + - 202 files modified, 6780+ insertions, 3961 deletions + +2. **Impact Analysis** + - Effects on activity development + - Adapter development implications + - Testing improvements + +3. **Migration Strategy Phases** + - High-level phase organization + - Dependency order + - Validation steps + +## Current Implementation Status + +### Completed Work (jshook/refactorings1 branch) +This branch extends main with the following completed refactoring work: + +#### Priority 1: Activity Class Consolidation ✅ +- **Status**: COMPLETE +- **Location**: `nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/Activity.java` +- **Details**: + - Unified Activity class (767 lines) combines all SimpleActivity and StandardActivity functionality + - Removed legacy types: `SimpleActivity`, `StandardActivity`, `StandardActivityType` + - Updated all callers: `ActivityLoader`, `ActivityExecutor`, `CoreMotor`, async helpers, tests + - Full `mvn verify` completed without regressions + - Benefits: reduced duplication, clearer instantiation, smaller adapter surface area +- **Reference**: See `ACTIVITY_REFACTORING_PLAN.md` for detailed step tracking + +### Reference Implementation (jshook/nb523 branch) +This branch demonstrates the intended architectural changes and serves as a reference: + +#### Reference: Configuration System Migration +- **Status**: Reference implementation exists on jshook/nb523 +- **Key Patterns Demonstrated**: + - ActivityDef → ActivityConfig migration approach (commit 97e7b16d7) + - ActivityConfig extends NBConfiguration + - Implements NBConfigurable and NBReconfigurable interfaces + - Type-safe parameter access with built-in validation +- **Application to refactorings1**: + - Will be implemented fresh on refactorings1 branch + - Using nb523 as architectural reference + - Applying patterns incrementally with verification at each step + +### Not Yet Started on refactorings1 + +#### Priority 2: Operation Pipeline (Should Have) +**Dependencies**: Requires Phase 1 completion +1. OpResolution implementation +2. Resolver components (OpResolverBank, AdapterResolver, DispenserResolver, ParsedOpResolver) +3. ActivityWiring setup +4. Metrics separation (ActivityMetrics extraction) + +#### Priority 3: Enhancements (Nice to Have) +**Dependencies**: Requires Phase 2 completion +1. Operation caching (OpCache) +2. Space API alignment +3. Dry-run capabilities +4. Advanced diagnostics +5. Performance optimizations + +#### Priority 4: Polish (Can Proceed in Parallel) +**Dependencies**: Can proceed once Phase 1 is complete +1. Visual documentation +2. Migration guides +3. Style configurations +4. Cleanup and deprecation removal + +## Branch Relationship +``` +main (8e85019fe) + ├── jshook/refactorings1 (current working branch) + │ └── Activity consolidation complete + │ └── Next: Apply ActivityConfig migration fresh + │ + └── jshook/nb523 (reference implementation, not to be merged) + └── Shows complete architectural vision + └── Used as pattern reference for incremental changes +``` + +**Strategy**: +1. Use nb523 as architectural reference +2. Apply changes incrementally on refactorings1 +3. Verify at each step before proceeding +4. Build cleanly from completed Activity consolidation + +## Key Implementation Decisions + +### Agreed Upon Approaches: +1. **Single Activity Class:** ✅ Unanimous agreement on consolidation - COMPLETE +2. **NBConfiguration Base:** 🔄 Standard configuration approach - IN PROGRESS +3. **Resolver Pattern:** ⚪ For operation synthesis pipeline - PLANNED +4. **Component Separation:** ⚪ Metrics, wiring, lifecycle as separate concerns - PLANNED +5. **Lazy Initialization:** ⚪ For performance optimization - PLANNED + +### Areas Requiring Further Discussion: +1. **Backward Compatibility:** How much to preserve vs. clean break + - Current approach: Breaking changes acceptable with clear migration path +2. **Performance Benchmarks:** Specific metrics to track + - Verify no regressions via `mvn verify` and existing test suite +3. **Testing Strategy:** Unit vs. integration test balance + - Approach: Comprehensive test coverage maintained at each phase + +## Success Metrics + +### Quantitative (Measured at Each Phase): +- **Code reduction**: ✅ Phase 1: ~1,300 lines removed (target: 1,500+ total) +- **Test coverage**: ✅ Phase 1: Maintained (all tests pass) +- **Performance**: ✅ Phase 1: No regression in operation throughput +- **Build time**: ✅ Phase 1: Acceptable + +### Qualitative (Ongoing Assessment): +- ✅ Developer experience improvement (single Activity class) +- 🔄 Clearer architecture documentation (in progress) +- 🔄 Simplified onboarding for new contributors (improved with consolidation) +- ⚪ Better error messages and debugging (planned for Phase 1.3) + +## Risk Mitigation + +### High Risk Items: +1. **Breaking Changes** + - Mitigation: Comprehensive migration guide + - Fallback: Compatibility layer if needed + +2. **Performance Regression** + - Mitigation: Benchmark before/after each phase + - Fallback: Revert specific optimizations + +3. **Adapter Compatibility** + - Mitigation: Test with all adapters early + - Fallback: Adapter-specific compatibility fixes + +### Medium Risk Items: +1. **Pattern Translation from nb523** + - Risk: Reference implementation may not directly apply + - Mitigation: Understand patterns first, then apply incrementally + - Status: Will be addressed during Phase 1.2 + +2. **Testing Gaps** + - Mitigation: Test-driven development, full `mvn verify` at each phase + - Fallback: Extended testing before proceeding to next phase + +## Implementation Roadmap + +### Phase 1: Core Consolidation ✅ MOSTLY COMPLETE +**Goal**: Unify activity hierarchy and standardize configuration +**Verification**: All tests pass, no runtime regressions, cleaner adapter API + +#### Step 1.1: Activity Class Consolidation ✅ COMPLETE +- [x] Activity class consolidation (jshook/refactorings1) + - [x] Create unified Activity class (767 lines) + - [x] Remove SimpleActivity, StandardActivity, StandardActivityType + - [x] Update ActivityLoader, ActivityExecutor, CoreMotor + - [x] Update all test classes + - [x] Verify with full test suite (`mvn verify`) +- **Verification**: Tests pass, ~1,300 net lines removed + +#### Step 1.2: Configuration System Migration ⚪ NOT STARTED +- [ ] ActivityDef → ActivityConfig migration (fresh implementation on refactorings1) + - [ ] Create ActivityConfig class extending NBConfiguration (reference: nb523 commit 97e7b16d7) + - [ ] Update Activity class to use ActivityConfig instead of ActivityDef + - [ ] Update core components to use NBConfigurable pattern + - [ ] Update ActivityLoader and related components + - [ ] Update all ActivityDef references across codebase + - [ ] Update adapter implementations + - [ ] Remove ActivityDef class + - [ ] Run full test suite +- **Verification**: All adapters build, tests pass, configuration validated +- **Reference**: See jshook/nb523 for implementation patterns + +#### Step 1.3: Lifecycle and Error Handling (Future) +- [ ] Basic lifecycle management improvements +- [ ] Core error handling enhancements +- **Verification**: State transitions validated, error paths tested + +### Phase 2: Operation Pipeline - NOT STARTED +**Goal**: Refactor operation synthesis for clarity and extensibility +**Dependencies**: Phase 1 must be complete +**Verification**: Operation resolution is deterministic, circular dependencies detected + +- [ ] OpResolution implementation +- [ ] Resolver components (OpResolverBank, AdapterResolver, DispenserResolver, ParsedOpResolver) +- [ ] ActivityWiring setup +- [ ] Metrics separation (ActivityMetrics extraction) +- **Verification**: All existing workloads run unchanged, metrics maintained + +### Phase 3: Enhancements - NOT STARTED +**Goal**: Add performance and developer experience improvements +**Dependencies**: Phase 2 must be complete +**Verification**: No performance regressions, enhanced capabilities verified + +- [ ] Operation caching (OpCache) +- [ ] Space API alignment improvements +- [ ] Dry-run capabilities enhancement +- [ ] Advanced diagnostics +- [ ] Performance optimizations +- **Verification**: Benchmark comparisons, diagnostic output validated + +### Phase 4: Polish - NOT STARTED +**Goal**: Improve documentation and developer onboarding +**Dependencies**: Can proceed once Phase 1 is complete +**Verification**: Documentation accurate, migration path clear + +- [ ] Visual documentation updates +- [ ] Migration guides for users +- [ ] Style configurations finalization +- [ ] Cleanup and deprecation removal +- **Verification**: Documentation review, community feedback + +## Summary of Current State + +**What's Complete**: +- ✅ Activity hierarchy consolidation (jshook/refactorings1) + - Single unified Activity class replacing SimpleActivity, StandardActivity, and interface + - 767 lines of consolidated functionality + - All tests passing with `mvn verify` + - ~1,300 net lines removed from codebase + +**What's Next**: +- ⚪ ActivityDef → ActivityConfig migration (to be implemented on refactorings1) + - Reference implementation exists on nb523 branch (commit 97e7b16d7) + - Will be applied fresh on refactorings1 using nb523 as architectural guide + - Incremental approach with verification at each step + +**What's Not Started**: +- Operation pipeline refactoring (OpResolution, resolvers, wiring) +- Metrics separation (ActivityMetrics extraction) +- Advanced features (caching, diagnostics, dry-run enhancements) +- Documentation and polish + +**Branch Strategy**: +- `main`: Stable production branch (commit 8e85019fe) +- `jshook/refactorings1`: Current working branch with Activity consolidation complete +- `jshook/nb523`: Reference implementation showing architectural target (not to be merged) + +**Immediate Actions Needed**: +1. Implement ActivityConfig on refactorings1 (using nb523 as reference) +2. Migrate Activity class from ActivityDef to ActivityConfig +3. Update all callers incrementally with verification +4. Complete ActivityDef removal across all adapters +5. Run full test suite at each major step +6. Proceed with remaining Phase 1 work, then Phase 2 + +## Conclusion + +The refactoring represents a fundamental simplification of NoSQLBench's architecture. Phase 1 Step 1 (Activity consolidation) is complete on the `refactorings1` branch. The `nb523` branch provides a reference implementation of the complete architectural vision, which will be applied incrementally on `refactorings1` with proper verification at each step. + +The detailed implementation steps in refactorings.md, combined with the impact analysis from branch_changes.md and the design rationale from nb523-branch-analysis.md, provide a comprehensive blueprint for executing this architectural transformation successfully. + +**Key Success Factors**: +1. **Incremental approach**: Apply changes in small, verifiable steps +2. **Reference-driven**: Use nb523 as architectural guide, not merge source +3. **Test-driven**: Run `mvn verify` at each major milestone +4. **Dependency-aware**: Complete each phase before proceeding to next +5. **Flexibility**: Adjust implementation details based on findings while maintaining architectural goals diff --git a/devdocs/devguide/drivers/optemplate_api.md b/devdocs/devguide/drivers/optemplate_api.md index 25f2ea74f..9129216d4 100644 --- a/devdocs/devguide/drivers/optemplate_api.md +++ b/devdocs/devguide/drivers/optemplate_api.md @@ -60,7 +60,12 @@ is responsible for doing two critical steps: Much of the boilerplate for doing this is handled already by the default Activity implementation which is used by all drivers: the -_SimpleActivity_. +`Activity` class. The concrete implementation now exposes direct setters +such as `setInputDispenser(...)`, `setActionDispenser(...)`, and +`setMotorDispenser(...)`, replacing the older `*Delegate` helpers. When you +provide dispensers through these methods, NoSQLBench automatically applies +the current `ActivityDef` to any listeners, removing the need for manual +wiring in driver code. ## Op Mapping diff --git a/nb-apis/nb-api/src/test/java/io/nosqlbench/testextensions/ThreadLeakDetectorExtension.java b/nb-apis/nb-api/src/test/java/io/nosqlbench/testextensions/ThreadLeakDetectorExtension.java new file mode 100644 index 000000000..80e1f466c --- /dev/null +++ b/nb-apis/nb-api/src/test/java/io/nosqlbench/testextensions/ThreadLeakDetectorExtension.java @@ -0,0 +1,109 @@ +// src/test/java/com/example/testing/ThreadLeakDetectorExtension.java +package io.nosqlbench.testextensions; + +/* + * Copyright (c) nosqlbench + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + + +import org.junit.jupiter.api.extension.*; +import java.lang.management.ManagementFactory; +import java.time.Duration; +import java.util.*; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +public class ThreadLeakDetectorExtension implements BeforeEachCallback, AfterEachCallback { + + // Config via system properties or junit-platform.properties -> systemProperties.* + private static final Duration GRACE = Duration.ofMillis( + Long.getLong("threadleak.grace.ms", 300)); // wait after test + private static final Duration JOIN_TIMEOUT = Duration.ofMillis( + Long.getLong("threadleak.join.ms", 200)); // join timeout per thread + private static final boolean IGNORE_DAEMONS = Boolean.parseBoolean( + System.getProperty("threadleak.ignoreDaemons", "true")); + private static final Pattern IGNORE_NAME = Pattern.compile( + System.getProperty("threadleak.ignoreNameRegex", + // common infra/IDE/GC threads; tune for your stack + "^(Finalizer|Reference Handler|Signal Dispatcher|Common-Cleaner|.*ForkJoinPool.*|.*Timer.*|.*Attach Listener.*|.*Process reaper.*)$")); + private static final Pattern IGNORE_GROUP = Pattern.compile( + System.getProperty("threadleak.ignoreGroupRegex", "^(system|InnocuousThreadGroup)$")); + + private final ThreadMXBeanShim mx = new ThreadMXBeanShim(); + + private Set beforeIds; + + @Override + public void beforeEach(ExtensionContext ctx) { + beforeIds = liveThreadIds(); + } + + @Override + public void afterEach(ExtensionContext ctx) throws Exception { + // Small grace to let pools/tasks shut down + Thread.sleep(GRACE.toMillis()); + + Set after = liveThreads(); + + // Try to join stragglers briefly (best-effort) + for (Thread t : after) { + if (t.isAlive() && !t.isDaemon()) { + try { t.join(JOIN_TIMEOUT.toMillis()); } catch (InterruptedException ignored) {} + } + } + + // Recompute after join attempt + after = liveThreads(); + + List leaked = after.stream() + // newly appeared since beforeEach + .filter(t -> beforeIds == null || !beforeIds.contains(t.getId())) + // filter typical noise + .filter(t -> !(IGNORE_DAEMONS && t.isDaemon())) + .filter(t -> !IGNORE_NAME.matcher(t.getName()).matches()) + .filter(t -> !IGNORE_GROUP.matcher(t.getThreadGroup() != null ? t.getThreadGroup().getName() : "").matches()) + .collect(Collectors.toList()); + + if (!leaked.isEmpty()) { + String details = leaked.stream() + .map(t -> String.format("#%d '%s' group=%s daemon=%s state=%s", + t.getId(), t.getName(), + (t.getThreadGroup() != null ? t.getThreadGroup().getName() : ""), + t.isDaemon(), t.getState())) + .collect(Collectors.joining("\n")); + + throw new AssertionError("Detected leaked threads after test:\n" + details); + } + } + + private Set liveThreadIds() { + long[] ids = mx.getAllThreadIds(); + Set set = new HashSet<>(ids.length); + for (long id : ids) set.add(id); + return set; + } + + private Set liveThreads() { + // getAllStackTraces() is a handy snapshot of live Thread objects + return Thread.getAllStackTraces().keySet(); + } + + // Minimal shim to avoid depending on com.sun.* at compile time + static final class ThreadMXBeanShim { + final java.lang.management.ThreadMXBean bean = ManagementFactory.getThreadMXBean(); + long[] getAllThreadIds() { return bean.getAllThreadIds(); } + } +} diff --git a/nb-apis/nb-api/src/test/java/io/nosqlbench/testextensions/ThreadLeakDetectorExtensionTest.java b/nb-apis/nb-api/src/test/java/io/nosqlbench/testextensions/ThreadLeakDetectorExtensionTest.java new file mode 100644 index 000000000..9044434b0 --- /dev/null +++ b/nb-apis/nb-api/src/test/java/io/nosqlbench/testextensions/ThreadLeakDetectorExtensionTest.java @@ -0,0 +1,251 @@ +/* + * Copyright (c) nosqlbench + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.nosqlbench.testextensions; + +import org.junit.jupiter.api.*; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Test cases demonstrating the ThreadLeakDetectorExtension functionality. + * This class shows both positive cases (proper cleanup) and negative cases (thread leaks). + */ +@ExtendWith(ThreadLeakDetectorExtension.class) +@TestMethodOrder(MethodOrderer.OrderAnnotation.class) +public class ThreadLeakDetectorExtensionTest { + + /** + * Test case that properly cleans up threads - should pass + */ + @Test + @Order(1) + public void testProperThreadCleanup() throws InterruptedException { + // Create an executor service + ExecutorService executor = Executors.newFixedThreadPool(2); + + // Submit some tasks + CountDownLatch latch = new CountDownLatch(2); + executor.submit(() -> { + try { + Thread.sleep(50); + latch.countDown(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + executor.submit(() -> { + try { + Thread.sleep(50); + latch.countDown(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + // Wait for tasks to complete + assertTrue(latch.await(1, TimeUnit.SECONDS), "Tasks should complete"); + + // Properly shutdown the executor + executor.shutdown(); + assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS), + "Executor should terminate"); + + // This test should pass as all threads are properly cleaned up + } + + /** + * Test case with daemon threads - should pass as daemon threads are ignored + */ + @Test + @Order(2) + public void testDaemonThreadsIgnored() { + // Create a daemon thread + Thread daemonThread = new Thread(() -> { + try { + // Daemon thread that runs briefly + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, "TestDaemonThread"); + + daemonThread.setDaemon(true); + daemonThread.start(); + + // Test passes even though daemon thread might still be running + // because ThreadLeakDetectorExtension ignores daemon threads by default + } + + /** + * Test case that creates a thread but properly cleans it up + */ + @Test + @Order(3) + public void testManualThreadCleanup() throws InterruptedException { + AtomicBoolean shouldRun = new AtomicBoolean(true); + + Thread worker = new Thread(() -> { + while (shouldRun.get()) { + try { + Thread.sleep(10); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + }, "ManualWorkerThread"); + + worker.start(); + + // Do some work + Thread.sleep(50); + + // Stop the thread properly + shouldRun.set(false); + worker.interrupt(); + worker.join(1000); + + assertFalse(worker.isAlive(), "Worker thread should be terminated"); + } + + /** + * Test case that would normally leak a thread but we'll disable the extension for this one + * This demonstrates what would happen without proper cleanup + */ + @Test + @Order(4) + @Disabled("This test intentionally leaks a thread - enable to see failure") + public void testIntentionalThreadLeak() { + // Create a thread that will outlive the test + Thread leakyThread = new Thread(() -> { + try { + // This thread will run for a long time + Thread.sleep(10000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, "LeakyThread"); + + leakyThread.start(); + + // Test ends without cleaning up the thread + // This would be detected by ThreadLeakDetectorExtension and fail + } + + /** + * Test case with thread pool that is not properly shutdown + */ + @Test + @Order(5) + @Disabled("This test intentionally leaks threads - enable to see failure") + public void testExecutorServiceLeak() { + // Create an executor but don't shut it down + ExecutorService executor = Executors.newFixedThreadPool(3); + + // Submit a task + executor.submit(() -> { + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + // Forgetting to call executor.shutdown() + // This would be detected as a thread leak + } + + /** + * Test case showing proper use of cached thread pool + */ + @Test + @Order(6) + public void testCachedThreadPoolCleanup() throws InterruptedException { + ExecutorService cachedPool = Executors.newCachedThreadPool(); + + // Submit multiple short tasks + for (int i = 0; i < 5; i++) { + final int taskId = i; + cachedPool.submit(() -> { + try { + Thread.sleep(20); + System.out.println("Task " + taskId + " completed"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + } + + // Properly shutdown and wait + cachedPool.shutdown(); + assertTrue(cachedPool.awaitTermination(2, TimeUnit.SECONDS), + "Cached pool should terminate"); + } + + /** + * Test that shows threads with special names are ignored + */ + @Test + @Order(7) + public void testIgnoredThreadPatterns() { + // Create a thread with a name that matches the ignore pattern + Thread ignoredThread = new Thread(() -> { + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, "ForkJoinPool-1-worker-1"); // This name matches the ignore pattern + + ignoredThread.start(); + + // Test passes even though thread might still be running + // because the name matches the ignore pattern in ThreadLeakDetectorExtension + } + + /** + * Nested test class to demonstrate extension works across test classes + */ + @Nested + @DisplayName("Nested Thread Tests") + class NestedThreadTests { + + @Test + public void testNestedThreadCleanup() throws InterruptedException { + Thread nestedWorker = new Thread(() -> { + try { + Thread.sleep(30); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, "NestedWorker"); + + nestedWorker.start(); + nestedWorker.join(1000); + + assertFalse(nestedWorker.isAlive(), "Nested worker should be terminated"); + } + } +} \ No newline at end of file diff --git a/nb-apis/nb-api/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/nb-apis/nb-api/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension new file mode 100644 index 000000000..7eeac0c59 --- /dev/null +++ b/nb-apis/nb-api/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension @@ -0,0 +1 @@ +io.nosqlbench.testextensions.ThreadLeakDetectorExtension diff --git a/nb-apis/nb-api/src/test/resources/junit-platform.properties b/nb-apis/nb-api/src/test/resources/junit-platform.properties new file mode 100644 index 000000000..fb4ee09e3 --- /dev/null +++ b/nb-apis/nb-api/src/test/resources/junit-platform.properties @@ -0,0 +1,2 @@ +# src/test/resources/junit-platform.properties +junit.jupiter.extensions.autodetection.enabled = true diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/Activity.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/Activity.java index e1eea06fd..0b7c51c38 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/Activity.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/Activity.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2022-2023 nosqlbench + * Copyright (c) 2022-2024 nosqlbench * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,137 +16,752 @@ package io.nosqlbench.engine.api.activityapi.core; -import io.nosqlbench.nb.api.components.core.NBComponent; -import io.nosqlbench.nb.api.engine.activityimpl.ActivityDef; -import io.nosqlbench.nb.api.engine.activityimpl.ParameterMap; +import io.nosqlbench.adapter.diag.DriverAdapterLoader; +import io.nosqlbench.adapters.api.activityconfig.OpsLoader; +import io.nosqlbench.adapters.api.activityconfig.yaml.OpTemplate; +import io.nosqlbench.adapters.api.activityconfig.yaml.OpTemplateFormat; +import io.nosqlbench.adapters.api.activityconfig.yaml.OpsDocList; +import io.nosqlbench.adapters.api.activityimpl.OpDispenser; +import io.nosqlbench.adapters.api.activityimpl.OpMapper; +import io.nosqlbench.adapters.api.activityimpl.uniform.DriverAdapter; +import io.nosqlbench.adapters.api.activityimpl.uniform.Space; +import io.nosqlbench.adapters.api.activityimpl.uniform.decorators.SyntheticOpTemplateProvider; +import io.nosqlbench.adapters.api.activityimpl.uniform.flowtypes.CycleOp; +import io.nosqlbench.adapters.api.templating.ParsedOp; +import io.nosqlbench.engine.api.activityimpl.uniform.ActivityDefAware; +import io.nosqlbench.engine.api.activityapi.core.progress.ActivityMetricProgressMeter; import io.nosqlbench.engine.api.activityapi.core.progress.ProgressCapable; +import io.nosqlbench.engine.api.activityapi.core.progress.ProgressMeterDisplay; import io.nosqlbench.engine.api.activityapi.core.progress.StateCapable; import io.nosqlbench.engine.api.activityapi.cyclelog.filters.IntPredicateDispenser; import io.nosqlbench.engine.api.activityapi.errorhandling.ErrorMetrics; +import io.nosqlbench.engine.api.activityapi.errorhandling.modular.NBErrorHandler; import io.nosqlbench.engine.api.activityapi.input.InputDispenser; import io.nosqlbench.engine.api.activityapi.output.OutputDispenser; -import io.nosqlbench.engine.api.activityapi.simrate.RateLimiter; -import io.nosqlbench.engine.api.activityimpl.SimpleActivity; +import io.nosqlbench.engine.api.activityapi.planning.OpSequence; +import io.nosqlbench.engine.api.activityapi.planning.SequencePlanner; +import io.nosqlbench.engine.api.activityapi.planning.SequencerType; +import io.nosqlbench.engine.api.activityapi.simrate.*; import io.nosqlbench.engine.api.activityimpl.motor.RunStateTally; +import io.nosqlbench.engine.api.activityimpl.OpFunctionComposition; +import io.nosqlbench.engine.api.activityimpl.Dryrun; +import io.nosqlbench.engine.core.lifecycle.scenario.container.InvokableResult; +import io.nosqlbench.nb.annotations.ServiceSelector; +import io.nosqlbench.nb.api.advisor.NBAdvisorOutput; +import io.nosqlbench.nb.api.advisor.NBAdvisorPoint; +import io.nosqlbench.nb.api.advisor.conditions.Conditions; +import io.nosqlbench.nb.api.components.core.NBComponent; +import io.nosqlbench.nb.api.components.events.NBEvent; +import io.nosqlbench.nb.api.components.events.ParamChange; +import io.nosqlbench.nb.api.components.events.SetThreads; +import io.nosqlbench.nb.api.components.status.NBStatusComponent; +import io.nosqlbench.nb.api.config.standard.*; +import io.nosqlbench.nb.api.engine.activityimpl.ActivityDef; +import io.nosqlbench.nb.api.engine.activityimpl.ParameterMap; +import io.nosqlbench.nb.api.engine.metrics.instruments.MetricCategory; +import io.nosqlbench.nb.api.errors.BasicError; +import io.nosqlbench.nb.api.errors.OpConfigError; +import io.nosqlbench.nb.api.labels.NBLabels; +import io.nosqlbench.nb.api.lifecycle.Shutdownable; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import java.io.InputStream; import java.io.PrintWriter; +import java.nio.charset.StandardCharsets; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.LongFunction; import java.util.function.Supplier; /** - * Provides the components needed to build and run an activity a runtime. - * The easiest way to build a useful Activity is to extend {@link SimpleActivity}. + * Unified Activity implementation combining SimpleActivity and StandardActivity functionality. + * This is the single concrete implementation of all activity functionality in NoSQLBench. */ -public interface Activity extends Comparable, ActivityDefObserver, ProgressCapable, StateCapable, NBComponent { +public class Activity extends NBStatusComponent + implements Comparable, ActivityDefObserver, ProgressCapable, StateCapable, + NBComponent, SyntheticOpTemplateProvider, InvokableResult { + + private static final Logger logger = LogManager.getLogger("ACTIVITY"); + + // Core fields from SimpleActivity + protected ActivityDef activityDef; + private final List closeables = new ArrayList<>(); + private final Map, Object> dispensers = new ConcurrentHashMap<>(); + private RunState runState = RunState.Uninitialized; + private ThreadLocal strideLimiterSource; + private ThreadLocal cycleLimiterSource; + private ActivityInstrumentation activityInstrumentation; + private PrintWriter console; + private long startedAtMillis; + private int nameEnumerator; + private ErrorMetrics errorMetrics; + private NBErrorHandler errorHandler; + private ActivityMetricProgressMeter progressMeter; + private String workloadSource = "unspecified"; + private final RunStateTally tally = new RunStateTally(); + + // Fields from StandardActivity + private OpSequence>> sequence; + private final ConcurrentHashMap, Space>> adapters = new ConcurrentHashMap<>(); /** - * Register an object which should be closed after this activity is shutdown. - * - * @param closeable An Autocloseable object + * Create a new Activity with the specified parent component and activity definition. + * This constructor handles both simple and standard activity initialization patterns. */ - void registerAutoCloseable(AutoCloseable closeable); + public Activity(NBComponent parent, ActivityDef activityDef) { + super(parent, NBLabels.forKV("activity", activityDef.getAlias()).and(activityDef.auxLabels())); + this.activityDef = activityDef; - ActivityDef getActivityDef(); + // Initialize alias if needed (from SimpleActivity) + if (activityDef.getAlias().equals(ActivityDef.DEFAULT_ALIAS)) { + Optional workloadOpt = activityDef.getParams().getOptionalString("workload", "yaml"); + if (workloadOpt.isPresent()) { + activityDef.getParams().set("alias", workloadOpt.get()); + } else { + activityDef.getParams().set("alias", + activityDef.getActivityDriver().toUpperCase(Locale.ROOT) + nameEnumerator); + nameEnumerator++; + } + } - default String getAlias() { + // Initialize StandardActivity components only if needed (when workload/ops are provided) + String defaultDriverName = activityDef.getActivityDriver(); + boolean hasWorkload = activityDef.getParams().getOptionalString("yaml", "workload").isPresent() || + activityDef.getParams().getOptionalString("op").isPresent() || + activityDef.getParams().getOptionalString("stmt", "statement").isPresent(); + + if (defaultDriverName != null && !defaultDriverName.isEmpty() && hasWorkload) { + initializeStandardComponents(defaultDriverName); + } + } + + public Activity(NBComponent parent, String activityDefString) { + this(parent, ActivityDef.parseActivityDef(activityDefString)); + } + + /** + * Initialize components needed for StandardActivity functionality. + * This is called when a driver is specified. + */ + private void initializeStandardComponents(String defaultDriverName) { + NBAdvisorPoint paramsAdvisor = create().advisor(b -> b.name("Workload")); + paramsAdvisor.add(Conditions.ValidNameError); + + List, Space>> adapterlist = new ArrayList<>(); + List pops = new ArrayList<>(); + ConcurrentHashMap, ? extends Space>> mappers = new ConcurrentHashMap<>(); + + NBConfigModel activityModel = activityDef.getConfigModel(); + NBConfigModel yamlmodel; + NBConfigModel adapterModel; + + DriverAdapter, Space> defaultAdapter = getDriverAdapter(defaultDriverName); + DriverAdapter, Space> adapter; + + OpsDocList workload; + Optional yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload"); + if (yaml_loc.isPresent()) { + Map disposable = new LinkedHashMap<>(activityDef.getParams()); + workload = loadStmtsDocList(); + yamlmodel = workload.getConfigModel(); + } else { + yamlmodel = ConfigModel.of(Activity.class).asReadOnly(); + } + yamlmodel.log(); + + NBConfigModel supersetConfig = ConfigModel.of(Activity.class).add(yamlmodel); + + // Load the op templates + List opTemplates = loadOpTemplates(defaultAdapter); + NBConfigModel combinedAdapterModel = ConfigModel.of(Activity.class); + + for (OpTemplate ot : opTemplates) { + logger.debug(() -> "Activity.opTemplate = " + ot); + String driverName = ot.getOptionalStringParam("driver", String.class) + .or(() -> ot.getOptionalStringParam("type", String.class)) + .orElse(defaultDriverName); + + if (!adapters.containsKey(driverName)) { + adapter = defaultDriverName.equals(driverName) ? defaultAdapter : getDriverAdapter(driverName); + NBConfigModel combinedModel = yamlmodel; + NBConfiguration combinedConfig = combinedModel.matchConfig(activityDef.getParams()); + + if (adapter instanceof NBConfigurable configurable) { + adapterModel = configurable.getConfigModel(); + combinedAdapterModel.add(adapterModel); + supersetConfig.add(adapterModel); + combinedModel = adapterModel.add(yamlmodel); + combinedConfig = combinedModel.matchConfig(activityDef.getParams()); + configurable.applyConfig(combinedConfig); + } + + adapters.put(driverName, adapter); + mappers.put(driverName, adapter.getOpMapper()); + } else { + adapter = adapters.get(driverName); + } + + paramsAdvisor.validateAll(ot.getParams().keySet()); + paramsAdvisor.validateAll(ot.getTags().keySet()); + paramsAdvisor.validateAll(ot.getBindings().keySet()); + adapterlist.add(adapter); + + ParsedOp pop = new ParsedOp(ot, adapter.getConfiguration(), List.of(adapter.getPreprocessor()), this); + logger.debug("Activity.pop=" + pop); + Optional discard = pop.takeOptionalStaticValue("driver", String.class); + pops.add(pop); + } + + logger.debug(() -> "Activity.opTemplate loop complete"); + + paramsAdvisor.setName("Workload", "Check parameters, template, and binding names") + .logName().evaluate(); + + paramsAdvisor.clear().setName("Superset", "Check overall parameters"); + paramsAdvisor.validateAll(supersetConfig.getNamedParams().keySet()); + paramsAdvisor.logName().evaluate(); + + combinedAdapterModel.assertNoConflicts(yamlmodel.getNamedParams(), "Template"); + combinedAdapterModel.log(); + supersetConfig.assertValidConfig(activityDef.getParams().getStringStringMap()); + supersetConfig.log(); + + if (0 == mappers.keySet().stream().filter(n -> n.equals(defaultDriverName)).count()) { + logger.warn(() -> "All op templates used a different driver than the default '" + defaultDriverName + "'"); + } + + try { + sequence = createOpSourceFromParsedOps(adapterlist, pops); + } catch (Exception e) { + if (e instanceof OpConfigError) { + throw e; + } + throw new OpConfigError("Error mapping workload template to operations: " + e.getMessage(), null, e); + } + + // Create metrics + create().gauge( + "ops_pending", + () -> this.getProgressMeter().getSummary().pending(), + MetricCategory.Core, + "The current number of operations which have not been dispatched for processing yet." + ); + create().gauge( + "ops_active", + () -> this.getProgressMeter().getSummary().current(), + MetricCategory.Core, + "The current number of operations which have been dispatched for processing, but which have not yet completed." + ); + create().gauge( + "ops_complete", + () -> this.getProgressMeter().getSummary().complete(), + MetricCategory.Core, + "The current number of operations which have been completed" + ); + } + + private DriverAdapter, Space> getDriverAdapter(String driverName) { + DriverAdapter, Space> adapter = Optional.of(driverName) + .flatMap(name -> ServiceSelector.of(name, ServiceLoader.load(DriverAdapterLoader.class)).get()) + .map(l -> l.load(this, NBLabels.forKV())) + .orElseThrow(() -> new OpConfigError("Unable to load '" + driverName + "' driver adapter.\n" + + "If this is a valid driver then you may need to rebuild NoSqlBench to include this driver adapter. " + + "Change 'false' for the driver in " + + "'./nb-adapters/pom.xml' and './nb-adapters/nb-adapters-included/pom.xml'.")); + + if (adapter instanceof ActivityDefAware aware) { + aware.setActivityDef(activityDef); + } + + return adapter; + } + + // Activity interface implementation methods + + public void registerAutoCloseable(AutoCloseable closeable) { + this.closeables.add(closeable); + } + + public ActivityDef getActivityDef() { + return activityDef; + } + + public String getAlias() { return this.getActivityDef().getAlias(); } - default ParameterMap getParams() { + public ParameterMap getParams() { return this.getActivityDef().getParams(); } - default void initActivity() { + public synchronized void initActivity() { + initOrUpdateRateLimiters(this.activityDef); + if (sequence != null) { + setDefaultsFromOpSequence(sequence); + } } - /** - * Close all autocloseables that have been registered with this Activity. - */ - void closeAutoCloseables(); - - MotorDispenser getMotorDispenserDelegate(); - - void setMotorDispenserDelegate(MotorDispenser motorDispenser); - - InputDispenser getInputDispenserDelegate(); - - void setInputDispenserDelegate(InputDispenser inputDispenser); - - ActionDispenser getActionDispenserDelegate(); - - void setActionDispenserDelegate(ActionDispenser actionDispenser); - - IntPredicateDispenser getResultFilterDispenserDelegate(); - - void setResultFilterDispenserDelegate(IntPredicateDispenser resultFilterDispenser); - - OutputDispenser getMarkerDispenserDelegate(); - - void setOutputDispenserDelegate(OutputDispenser outputDispenser); - - @Override - RunState getRunState(); - - void setRunState(RunState runState); - - long getStartedAtMillis(); - - default void shutdownActivity() { + public void closeAutoCloseables() { + for (AutoCloseable closeable : closeables) { + logger.debug(() -> "CLOSING " + closeable.getClass().getCanonicalName() + ": " + closeable); + try { + closeable.close(); + } catch (Exception e) { + throw new RuntimeException("Error closing " + closeable + ": " + e, e); + } + } + closeables.clear(); } - default String getCycleSummary() { + public MotorDispenser getMotorDispenser() { + return getDispenser(MotorDispenser.class); + } + + public void setMotorDispenser(MotorDispenser motorDispenser) { + setDispenser(MotorDispenser.class, motorDispenser); + } + + public InputDispenser getInputDispenser() { + return getDispenser(InputDispenser.class); + } + + public void setInputDispenser(InputDispenser inputDispenser) { + setDispenser(InputDispenser.class, inputDispenser); + } + + public ActionDispenser getActionDispenser() { + return getDispenser(ActionDispenser.class); + } + + public void setActionDispenser(ActionDispenser actionDispenser) { + setDispenser(ActionDispenser.class, actionDispenser); + } + + public IntPredicateDispenser getResultFilterDispenser() { + return getDispenser(IntPredicateDispenser.class); + } + + public void setResultFilterDispenser(IntPredicateDispenser resultFilterDispenser) { + setDispenser(IntPredicateDispenser.class, resultFilterDispenser); + } + + public OutputDispenser getOutputDispenser() { + return getDispenser(OutputDispenser.class); + } + + public void setOutputDispenser(OutputDispenser outputDispenser) { + setDispenser(OutputDispenser.class, outputDispenser); + } + + public T getDispenser(Class type) { + Object dispenser = dispensers.get(type); + return dispenser == null ? null : type.cast(dispenser); + } + + public void setDispenser(Class type, T dispenser) { + if (dispenser == null) { + dispensers.remove(type); + } else { + dispensers.put(type, dispenser); + ActivityDefObserver.apply(activityDef, dispenser); + } + } + + public synchronized RunState getRunState() { + return runState; + } + + public synchronized void setRunState(RunState runState) { + this.runState = runState; + if (RunState.Running == runState) { + this.startedAtMillis = System.currentTimeMillis(); + } + } + + public long getStartedAtMillis() { + return startedAtMillis; + } + + public void shutdownActivity() { + // Shutdown adapters from StandardActivity + for (Map.Entry, Space>> entry : adapters.entrySet()) { + String adapterName = entry.getKey(); + DriverAdapter adapter = entry.getValue(); + if (adapter instanceof AutoCloseable autoCloseable) { + try { + autoCloseable.close(); + } catch (Exception e) { + logger.error("Error closing adapter " + adapterName + ": " + e.getMessage(), e); + } + } + } + } + + public String getCycleSummary() { return this.getActivityDef().getCycleSummary(); } - /** - * Get the current cycle rate limiter for this activity. - * The cycle rate limiter is used to throttle the rate at which - * cycles are dispatched across all threads in the activity - * @return the cycle {@link RateLimiter} - */ - RateLimiter getCycleLimiter(); + public RateLimiter getCycleLimiter() { + if (cycleLimiterSource != null) { + return cycleLimiterSource.get(); + } else { + return null; + } + } + public synchronized RateLimiter getStrideLimiter() { + if (strideLimiterSource != null) { + return strideLimiterSource.get(); + } else { + return null; + } + } - /** - * Get the current stride rate limiter for this activity. - * The stride rate limiter is used to throttle the rate at which - * new strides are dispatched across all threads in an activity. - * @return The stride {@link RateLimiter} - */ - RateLimiter getStrideLimiter(); + public synchronized ActivityInstrumentation getInstrumentation() { + if (null == this.activityInstrumentation) { + activityInstrumentation = new ComponentActivityInstrumentation(this); + } + return activityInstrumentation; + } - /** - * Get or create the instrumentation needed for this activity. This provides - * a single place to find and manage, and document instrumentation that is - * uniform across all activities. - * - * @return A new or existing instrumentation object for this activity. - */ - ActivityInstrumentation getInstrumentation(); + public synchronized PrintWriter getConsoleOut() { + if (null == console) { + this.console = new PrintWriter(System.out, false, StandardCharsets.UTF_8); + } + return this.console; + } - PrintWriter getConsoleOut(); + public synchronized InputStream getConsoleIn() { + return System.in; + } - InputStream getConsoleIn(); + public void setConsoleOut(PrintWriter writer) { + this.console = writer; + } - void setConsoleOut(PrintWriter writer); + public synchronized ErrorMetrics getExceptionMetrics() { + if (null == this.errorMetrics) { + errorMetrics = new ErrorMetrics(this); + } + return errorMetrics; + } - ErrorMetrics getExceptionMetrics(); + public int getMaxTries() { + return this.activityDef.getParams().getOptionalInteger("maxtries").orElse(10); + } -// /** -// * When a driver needs to identify an error uniquely for the purposes of -// * routing it to the correct error handler, or naming it in logs, or naming -// * metrics, override this method in your activity. -// * @return A function that can reliably and safely map an instance of Throwable to a stable name. -// */ -// default Function getErrorNameMapper() { -// return t -> t.getClass().getSimpleName(); -// } -// - int getMaxTries(); - - default int getHdrDigits() { + public int getHdrDigits() { return this.getParams().getOptionalInteger("hdr_digits").orElse(4); } - RunStateTally getRunStateTally(); + public RunStateTally getRunStateTally() { + return tally; + } + + public int compareTo(Activity o) { + return getAlias().compareTo(o.getAlias()); + } + + public synchronized void onActivityDefUpdate(ActivityDef activityDef) { + initOrUpdateRateLimiters(activityDef); + this.errorHandler = null; + + for (DriverAdapter adapter : adapters.values()) { + if (adapter instanceof NBReconfigurable configurable) { + NBConfigModel cfgModel = configurable.getReconfigModel(); + NBConfiguration cfg = cfgModel.matchConfig(activityDef.getParams()); + NBReconfigurable.applyMatching(cfg, List.of(configurable)); + } + } + } + + public synchronized ProgressMeterDisplay getProgressMeter() { + if (null == this.progressMeter) { + this.progressMeter = new ActivityMetricProgressMeter((Activity) this); + } + return this.progressMeter; + } + + public Map asResult() { + return Map.of("activity", this.getAlias()); + } + + @Override + public List getSyntheticOpTemplates(OpsDocList opsDocList, Map cfg) { + List opTemplates = new ArrayList<>(); + for (DriverAdapter adapter : adapters.values()) { + if (adapter instanceof SyntheticOpTemplateProvider sotp) { + List newTemplates = sotp.getSyntheticOpTemplates(opsDocList, cfg); + opTemplates.addAll(newTemplates); + } + } + return opTemplates; + } + + @Override + public void onEvent(NBEvent event) { + switch (event) { + case ParamChange pc -> { + switch (pc.value()) { + case SetThreads st -> activityDef.setThreads(st.threads); + case CycleRateSpec crs -> createOrUpdateCycleLimiter(crs); + case StrideRateSpec srs -> createOrUpdateStrideLimiter(srs); + default -> super.onEvent(event); + } + } + default -> super.onEvent(event); + } + } + + // Helper methods from SimpleActivity + + public synchronized NBErrorHandler getErrorHandler() { + if (null == this.errorHandler) { + errorHandler = new NBErrorHandler( + () -> activityDef.getParams().getOptionalString("errors").orElse("stop"), + this::getExceptionMetrics); + } + return errorHandler; + } + + public synchronized void initOrUpdateRateLimiters(ActivityDef activityDef) { + activityDef.getParams().getOptionalNamedParameter("striderate") + .map(StrideRateSpec::new) + .ifPresentOrElse( + sr -> this.onEvent(new ParamChange<>(sr)), + () -> strideLimiterSource = null + ); + + activityDef.getParams().getOptionalNamedParameter("cyclerate", "targetrate", "rate") + .map(CycleRateSpec::new) + .ifPresentOrElse( + sr -> this.onEvent(new ParamChange<>(sr)), + () -> cycleLimiterSource = null + ); + } + + public void createOrUpdateStrideLimiter(SimRateSpec spec) { + strideLimiterSource = ThreadLocalRateLimiters.createOrUpdate(this, strideLimiterSource, spec); + } + + public void createOrUpdateCycleLimiter(SimRateSpec spec) { + cycleLimiterSource = ThreadLocalRateLimiters.createOrUpdate(this, cycleLimiterSource, spec); + } + + public synchronized void setDefaultsFromOpSequence(OpSequence seq) { + Optional strideOpt = getParams().getOptionalString("stride"); + if (strideOpt.isEmpty()) { + String stride = String.valueOf(seq.getSequence().length); + logger.info(() -> "defaulting stride to " + stride + " (the sequence length)"); + getParams().setSilently("stride", stride); + } + + // CYCLES + Optional cyclesOpt = getParams().getOptionalString("cycles"); + if (cyclesOpt.isEmpty()) { + String cycles = getParams().getOptionalString("stride").orElseThrow(); + logger.info(() -> "defaulting cycles to " + cycles + " (the stride length)"); + this.getActivityDef().setCycles(getParams().getOptionalString("stride").orElseThrow()); + } else { + if (0 == activityDef.getCycleCount()) { + throw new RuntimeException( + "You specified cycles, but the range specified means zero cycles: " + getParams().get("cycles") + ); + } + long stride = getParams().getOptionalLong("stride").orElseThrow(); + long cycles = this.activityDef.getCycleCount(); + if (cycles < stride) { + throw new RuntimeException( + "The specified cycles (" + cycles + ") are less than the stride (" + stride + "). This means there aren't enough cycles to cause a stride to be executed." + + " If this was intended, then set stride low enough to allow it." + ); + } + } + + long cycleCount = this.activityDef.getCycleCount(); + long stride = this.activityDef.getParams().getOptionalLong("stride").orElseThrow(); + + if (0 < stride && 0 != cycleCount % stride) { + logger.warn(() -> "The stride does not evenly divide cycles. Only full strides will be executed," + + "leaving some cycles unused. (stride=" + stride + ", cycles=" + cycleCount + ')'); + } + + Optional threadSpec = activityDef.getParams().getOptionalString("threads"); + if (threadSpec.isPresent()) { + String spec = threadSpec.get(); + int processors = Runtime.getRuntime().availableProcessors(); + if ("auto".equalsIgnoreCase(spec)) { + int threads = processors * 10; + if (threads > activityDef.getCycleCount()) { + threads = (int) activityDef.getCycleCount(); + logger.info("setting threads to {} (auto) [10xCORES, cycle count limited]", threads); + } else { + logger.info("setting threads to {} (auto) [10xCORES]", threads); + } + activityDef.getParams().setSilently("threads", threads); + } else if (spec.toLowerCase().matches("\\d+x")) { + String multiplier = spec.substring(0, spec.length() - 1); + int threads = processors * Integer.parseInt(multiplier); + logger.info(() -> "setting threads to " + threads + " (" + multiplier + "x)"); + activityDef.getParams().setSilently("threads", threads); + } else if (spec.toLowerCase().matches("\\d+")) { + logger.info(() -> "setting threads to " + spec + " (direct)"); + activityDef.getParams().setSilently("threads", Integer.parseInt(spec)); + } + + if (activityDef.getThreads() > activityDef.getCycleCount()) { + logger.warn(() -> "threads=" + activityDef.getThreads() + " and cycles=" + activityDef.getCycleSummary() + + ", you should have more cycles than threads."); + } + + } else if (1000 < cycleCount) { + logger.warn(() -> "For testing at scale, it is highly recommended that you " + + "set threads to a value higher than the default of 1." + + " hint: you can use threads=auto for reasonable default, or" + + " consult the topic on threads with `help threads` for" + + " more information."); + } + + if (0 < this.activityDef.getCycleCount() && seq.getOps().isEmpty()) { + throw new BasicError("You have configured a zero-length sequence and non-zero cycles. It is not possible to continue with this activity."); + } + } + + protected OpSequence>> createOpSourceFromParsedOps( + List, Space>> adapters, + List pops + ) { + try { + List ratios = new ArrayList<>(pops.size()); + + for (ParsedOp pop : pops) { + long ratio = pop.takeStaticConfigOr("ratio", 1); + ratios.add(ratio); + } + + SequencerType sequencerType = getParams() + .getOptionalString("seq") + .map(SequencerType::valueOf) + .orElse(SequencerType.bucket); + SequencePlanner>> planner = new SequencePlanner<>(sequencerType); + + for (int i = 0; i < pops.size(); i++) { + long ratio = ratios.get(i); + ParsedOp pop = pops.get(i); + + try { + if (0 == ratio) { + logger.info(() -> "skipped mapping op '" + pop.getName() + '\''); + continue; + } + + DriverAdapter, Space> adapter = adapters.get(i); + OpMapper, Space> opMapper = adapter.getOpMapper(); + LongFunction spaceFunc = adapter.getSpaceFunc(pop); + OpDispenser> dispenser = opMapper.apply(this, pop, spaceFunc); + String dryrunSpec = pop.takeStaticConfigOr("dryrun", "none"); + Dryrun dryrun = pop.takeEnumFromFieldOr(Dryrun.class, Dryrun.none, "dryrun"); + dispenser = OpFunctionComposition.wrapOptionally(adapter, dispenser, pop, dryrun); + + planner.addOp((OpDispenser>) dispenser, ratio); + } catch (Exception e) { + throw new OpConfigError("Error while mapping op from template named '" + pop.getName() + "': " + e.getMessage(), e); + } + } + + return planner.resolve(); + + } catch (Exception e) { + if (e instanceof OpConfigError oce) { + throw oce; + } else { + throw new OpConfigError(e.getMessage(), workloadSource, e); + } + } + } + + protected List loadOpTemplates(DriverAdapter defaultDriverAdapter) { + String tagfilter = activityDef.getParams().getOptionalString("tags").orElse(""); + + OpsDocList opsDocList = loadStmtsDocList(); + + List unfilteredOps = opsDocList.getOps(false); + List filteredOps = opsDocList.getOps(tagfilter, true); + + if (filteredOps.isEmpty()) { + if (!unfilteredOps.isEmpty()) { + String message = "There were no active op templates with tag filter '" + tagfilter + "', since all " + + unfilteredOps.size() + " were filtered out. Examine the session log for details"; + NBAdvisorOutput.test(message); + } + if (defaultDriverAdapter instanceof SyntheticOpTemplateProvider sotp) { + filteredOps = sotp.getSyntheticOpTemplates(opsDocList, this.activityDef.getParams()); + Objects.requireNonNull(filteredOps); + if (filteredOps.isEmpty()) { + throw new BasicError("Attempted to create synthetic ops from driver '" + defaultDriverAdapter.getAdapterName() + '\'' + + " but no ops were created. You must provide either a workload or an op parameter. Activities require op templates."); + } + } else { + throw new BasicError(""" + No op templates were provided. You must provide one of these activity parameters: + 1) workload=some.yaml + 2) op='inline template' + 3) driver=stdout (or any other drive that can synthesize ops)"""); + } + } + + return filteredOps; + } + + protected OpsDocList loadStmtsDocList() { + try { + String op = activityDef.getParams().getOptionalString("op").orElse(null); + String stmt = activityDef.getParams().getOptionalString("stmt", "statement").orElse(null); + String workload = activityDef.getParams().getOptionalString("workload").orElse(null); + + if ((op != null ? 1 : 0) + (stmt != null ? 1 : 0) + (workload != null ? 1 : 0) > 1) { + throw new OpConfigError("Only op, statement, or workload may be provided, not more than one."); + } + + logger.debug("loadStmtsDocList #1"); + if (workload != null && OpsLoader.isJson(workload)) { + workloadSource = "commandline: (workload/json):" + workload; + return OpsLoader.loadString(workload, OpTemplateFormat.json, activityDef.getParams(), null); + } else if (workload != null && OpsLoader.isYaml(workload)) { + workloadSource = "commandline: (workload/yaml):" + workload; + return OpsLoader.loadString(workload, OpTemplateFormat.yaml, activityDef.getParams(), null); + } else if (workload != null) { + return OpsLoader.loadPath(workload, activityDef.getParams(), "activities"); + } + + logger.debug("loadStmtsDocList #2"); + if (stmt != null) { + workloadSource = "commandline: (stmt/inline): '" + stmt + "'"; + return OpsLoader.loadString(stmt, OpTemplateFormat.inline, activityDef.getParams(), null); + } + + logger.debug("loadStmtsDocList #3"); + if (op != null && OpsLoader.isJson(op)) { + workloadSource = "commandline: (op/json): '" + op + "'"; + return OpsLoader.loadString(op, OpTemplateFormat.json, activityDef.getParams(), null); + } else if (op != null) { + workloadSource = "commandline: (op/inline): '" + op + "'"; + return OpsLoader.loadString(op, OpTemplateFormat.inline, activityDef.getParams(), null); + } + + return OpsDocList.none(); + + } catch (Exception e) { + throw new OpConfigError("Error loading op templates: " + e, workloadSource, e); + } + } + + public OpSequence>> getOpSequence() { + return sequence; + } + + public String toString() { + return (activityDef != null ? activityDef.getAlias() : "unset_alias") + ':' + this.runState + ':' + this.tally; + } } diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/BaseAsyncAction.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/BaseAsyncAction.java index b3fe09394..47747453e 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/BaseAsyncAction.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/core/BaseAsyncAction.java @@ -25,17 +25,16 @@ import org.apache.logging.log4j.LogManager; /** * * @param An type of state holder for an operation, holding everything unique to that cycle and operation - * @param An type of of an Activity, a state holder for a runtime instance of an Activity */ -public abstract class BaseAsyncAction implements AsyncAction, Stoppable, ActivityDefObserver { +public abstract class BaseAsyncAction implements AsyncAction, Stoppable, ActivityDefObserver { private final static Logger logger = LogManager.getLogger("BaseAsyncAction"); - protected final A activity; + protected final Activity activity; protected int slot; protected boolean running = true; - public BaseAsyncAction(A activity, int slot) { + public BaseAsyncAction(Activity activity, int slot) { this.activity = activity; this.slot = slot; diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/cyclelog/outputs/cyclelog/CycleLogOutputType.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/cyclelog/outputs/cyclelog/CycleLogOutputType.java index 65061c3ec..b6d18e3aa 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/cyclelog/outputs/cyclelog/CycleLogOutputType.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityapi/cyclelog/outputs/cyclelog/CycleLogOutputType.java @@ -42,7 +42,7 @@ public class CycleLogOutputType implements OutputType { public Dispenser(Activity activity) { this.activity = activity; - Input input = activity.getInputDispenserDelegate().getInput(0); + Input input = activity.getInputDispenser().getInput(0); CycleLogOutput rleFileWriter = new CycleLogOutput(activity); // TODO: Rework this so that the contiguous marking chunker can onAfterOpStop filtering diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/CoreServices.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/CoreServices.java index a6c8f218d..58ef5d83f 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/CoreServices.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/CoreServices.java @@ -31,57 +31,40 @@ import java.util.function.Predicate; public class CoreServices { - public static Optional getOutputDispenser(A activity) { + public static Optional getOutputDispenser(Activity activity) { OutputDispenser outputDispenser = new SimpleConfig(activity, "output").getString("type") - .flatMap(OutputType.FINDER::get) - .map(mt -> mt.getOutputDispenser(activity)).orElse(null); - if (outputDispenser==null) { + .flatMap(OutputType.FINDER::get) + .map(mt -> mt.getOutputDispenser(activity)) + .orElse(null); + + if (outputDispenser == null) { return Optional.empty(); } - Optional> outputFilterDispenser = getOutputFilter(activity); - if (outputFilterDispenser.isPresent()) { - outputDispenser = new FilteringOutputDispenser(outputDispenser, outputFilterDispenser.get()); - } - - return Optional.ofNullable(outputDispenser); + return Optional.of(applyOutputFilter(outputDispenser, resolveFilter(activity, "of", "outputfilter"))); } - public static Optional> getOutputFilter(A activity) { - String paramdata= activity.getParams().getOptionalString("of") - .orElse(activity.getParams().getOptionalString("outputfilter").orElse(null)); - if (paramdata==null) { - return Optional.empty(); - } - return getFilterPredicate(paramdata); - } - -// public static Optional getResultFilterDispenser(A activity) { -// Optional intPredicateDispenser = new SimpleConfig(activity, "resultfilter") -// .getString("type") -// .flatMap(ExperimentalResultFilterType.FINDER::get) -// .map(rft -> rft.getFilterDispenser(activity)); -// return intPredicateDispenser; -// } -// - public static InputDispenser getInputDispenser(A activity) { + public static InputDispenser getInputDispenser(Activity activity) { String inputTypeName = new SimpleConfig(activity, "input").getString("type").orElse("atomicseq"); InputType inputType = InputType.FINDER.getOrThrow(inputTypeName); InputDispenser dispenser = inputType.getInputDispenser(activity); - Optional> inputFilterDispenser = getInputFilter(activity); - if (inputFilterDispenser.isPresent()) { - dispenser = new FilteringInputDispenser(dispenser, inputFilterDispenser.get()); + Optional> inputFilter = resolveFilter(activity, "if", "inputfilter"); + if (inputFilter.isPresent()) { + dispenser = new FilteringInputDispenser(dispenser, inputFilter.get()); } return dispenser; } - public static Optional> getInputFilter(A activity) { - String paramdata= activity.getParams().getOptionalString("if") - .orElse(activity.getParams().getOptionalString("inputfilter").orElse(null)); - if (paramdata==null) { - return Optional.empty(); - } - return getFilterPredicate(paramdata); + private static OutputDispenser applyOutputFilter(OutputDispenser dispenser, Optional> filter) { + return filter + .map(predicate -> (OutputDispenser) new FilteringOutputDispenser(dispenser, predicate)) + .orElse(dispenser); + } + + private static Optional> resolveFilter(Activity activity, String shortKey, String longKey) { + return activity.getParams().getOptionalString(shortKey) + .or(() -> activity.getParams().getOptionalString(longKey)) + .flatMap(CoreServices::getFilterPredicate); } private static Optional> getFilterPredicate(String paramdata) { diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/SimpleActivity.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/SimpleActivity.java deleted file mode 100644 index e656e9455..000000000 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/SimpleActivity.java +++ /dev/null @@ -1,669 +0,0 @@ -/* - * Copyright (c) 2022-2024 nosqlbench - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package io.nosqlbench.engine.api.activityimpl; - -import io.nosqlbench.adapters.api.activityconfig.OpsLoader; -import io.nosqlbench.adapters.api.activityconfig.yaml.OpTemplate; -import io.nosqlbench.adapters.api.activityconfig.yaml.OpTemplateFormat; -import io.nosqlbench.adapters.api.activityconfig.yaml.OpsDocList; -import io.nosqlbench.adapters.api.activityimpl.OpDispenser; -import io.nosqlbench.adapters.api.activityimpl.OpMapper; -import io.nosqlbench.adapters.api.activityimpl.uniform.DriverAdapter; -import io.nosqlbench.adapters.api.activityimpl.uniform.Space; -import io.nosqlbench.adapters.api.activityimpl.uniform.decorators.SyntheticOpTemplateProvider; -import io.nosqlbench.adapters.api.activityimpl.uniform.flowtypes.CycleOp; -import io.nosqlbench.adapters.api.templating.ParsedOp; -import io.nosqlbench.engine.api.activityapi.core.*; -import io.nosqlbench.engine.api.activityapi.core.progress.ActivityMetricProgressMeter; -import io.nosqlbench.engine.api.activityapi.core.progress.ProgressMeterDisplay; -import io.nosqlbench.engine.api.activityapi.cyclelog.filters.IntPredicateDispenser; -import io.nosqlbench.engine.api.activityapi.errorhandling.ErrorMetrics; -import io.nosqlbench.engine.api.activityapi.errorhandling.modular.NBErrorHandler; -import io.nosqlbench.engine.api.activityapi.input.InputDispenser; -import io.nosqlbench.engine.api.activityapi.output.OutputDispenser; -import io.nosqlbench.engine.api.activityapi.planning.OpSequence; -import io.nosqlbench.engine.api.activityapi.planning.SequencePlanner; -import io.nosqlbench.engine.api.activityapi.planning.SequencerType; -import io.nosqlbench.engine.api.activityapi.simrate.*; -import io.nosqlbench.engine.api.activityimpl.motor.RunStateTally; -import io.nosqlbench.engine.core.lifecycle.scenario.container.InvokableResult; -import io.nosqlbench.nb.api.advisor.NBAdvisorOutput; -import io.nosqlbench.nb.api.components.core.NBComponent; -import io.nosqlbench.nb.api.components.events.ParamChange; -import io.nosqlbench.nb.api.components.status.NBStatusComponent; -import io.nosqlbench.nb.api.engine.activityimpl.ActivityDef; -import io.nosqlbench.nb.api.errors.BasicError; -import io.nosqlbench.nb.api.errors.OpConfigError; -import io.nosqlbench.nb.api.labels.NBLabels; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; - -import java.io.InputStream; -import java.io.PrintWriter; -import java.nio.charset.StandardCharsets; -import java.util.*; -import java.util.function.Function; -import java.util.function.LongFunction; - -/** - * A default implementation of an Activity, suitable for building upon. - */ -public class SimpleActivity extends NBStatusComponent implements Activity, InvokableResult { - private static final Logger logger = LogManager.getLogger("ACTIVITY"); - - protected ActivityDef activityDef; - private final List closeables = new ArrayList<>(); - private MotorDispenser motorDispenser; - private InputDispenser inputDispenser; - private ActionDispenser actionDispenser; - private OutputDispenser markerDispenser; - private IntPredicateDispenser resultFilterDispenser; - private RunState runState = RunState.Uninitialized; - private ThreadLocal strideLimiterSource; - private ThreadLocal cycleLimiterSource; - private ActivityInstrumentation activityInstrumentation; - private PrintWriter console; - private long startedAtMillis; - private int nameEnumerator; - private ErrorMetrics errorMetrics; - private NBErrorHandler errorHandler; - private ActivityMetricProgressMeter progressMeter; - private String workloadSource = "unspecified"; - private final RunStateTally tally = new RunStateTally(); - - public SimpleActivity(NBComponent parent, ActivityDef activityDef) { - super(parent, NBLabels.forKV("activity", activityDef.getAlias()).and(activityDef.auxLabels())); - this.activityDef = activityDef; - if (activityDef.getAlias().equals(ActivityDef.DEFAULT_ALIAS)) { - Optional workloadOpt = activityDef.getParams().getOptionalString( - "workload", - "yaml" - ); - if (workloadOpt.isPresent()) { - activityDef.getParams().set("alias", workloadOpt.get()); - } else { - activityDef.getParams().set("alias", - activityDef.getActivityDriver().toUpperCase(Locale.ROOT) - + nameEnumerator); - nameEnumerator++; - } - } - } - - public SimpleActivity(NBComponent parent, String activityDefString) { - this(parent, ActivityDef.parseActivityDef(activityDefString)); - } - - @Override - public synchronized void initActivity() { - initOrUpdateRateLimiters(this.activityDef); - } - - public synchronized NBErrorHandler getErrorHandler() { - if (null == this.errorHandler) { - errorHandler = new NBErrorHandler( - () -> activityDef.getParams().getOptionalString("errors").orElse("stop"), - this::getExceptionMetrics); - } - return errorHandler; - } - - @Override - public synchronized RunState getRunState() { - return runState; - } - - @Override - public synchronized void setRunState(RunState runState) { - this.runState = runState; - if (RunState.Running == runState) { - this.startedAtMillis = System.currentTimeMillis(); - } - } - - @Override - public long getStartedAtMillis() { - return startedAtMillis; - } - - @Override - public final MotorDispenser getMotorDispenserDelegate() { - return motorDispenser; - } - - @Override - public final void setMotorDispenserDelegate(MotorDispenser motorDispenser) { - this.motorDispenser = motorDispenser; - } - - @Override - public final InputDispenser getInputDispenserDelegate() { - return inputDispenser; - } - - @Override - public final void setInputDispenserDelegate(InputDispenser inputDispenser) { - this.inputDispenser = inputDispenser; - } - - @Override - public final ActionDispenser getActionDispenserDelegate() { - return actionDispenser; - } - - @Override - public final void setActionDispenserDelegate(ActionDispenser actionDispenser) { - this.actionDispenser = actionDispenser; - } - - @Override - public IntPredicateDispenser getResultFilterDispenserDelegate() { - return resultFilterDispenser; - } - - @Override - public void setResultFilterDispenserDelegate(IntPredicateDispenser resultFilterDispenser) { - this.resultFilterDispenser = resultFilterDispenser; - } - - @Override - public OutputDispenser getMarkerDispenserDelegate() { - return this.markerDispenser; - } - - @Override - public void setOutputDispenserDelegate(OutputDispenser outputDispenser) { - this.markerDispenser = outputDispenser; - } - - @Override - public ActivityDef getActivityDef() { - return activityDef; - } - - public String toString() { - return (activityDef != null ? activityDef.getAlias() : "unset_alias") + ':' + this.runState + ':' + this.tally; - } - - @Override - public int compareTo(Activity o) { - return getAlias().compareTo(o.getAlias()); - } - - @Override - public void registerAutoCloseable(AutoCloseable closeable) { - this.closeables.add(closeable); - } - - @Override - public void closeAutoCloseables() { - for (AutoCloseable closeable : closeables) { - logger.debug(() -> "CLOSING " + closeable.getClass().getCanonicalName() + ": " + closeable); - try { - closeable.close(); - } catch (Exception e) { - throw new RuntimeException("Error closing " + closeable + ": " + e, e); - } - } - closeables.clear(); - } - - @Override - public RateLimiter getCycleLimiter() { - if (cycleLimiterSource!=null) { - return cycleLimiterSource.get(); - } else { - return null; - } - } - @Override - public synchronized RateLimiter getStrideLimiter() { - if (strideLimiterSource!=null) { - return strideLimiterSource.get(); - } else { - return null; - } - } - - @Override - public synchronized ActivityInstrumentation getInstrumentation() { - if (null == this.activityInstrumentation) { - activityInstrumentation = new ComponentActivityInstrumentation(this); -// activityInstrumentation = new CoreActivityInstrumentation(this); - } - return activityInstrumentation; - } - - @Override - public synchronized PrintWriter getConsoleOut() { - if (null == console) { - this.console = new PrintWriter(System.out, false, StandardCharsets.UTF_8); - } - return this.console; - } - - @Override - public synchronized InputStream getConsoleIn() { - return System.in; - } - - @Override - public void setConsoleOut(PrintWriter writer) { - this.console = writer; - } - - @Override - public synchronized ErrorMetrics getExceptionMetrics() { - if (null == this.errorMetrics) { - errorMetrics = new ErrorMetrics(this); - } - return errorMetrics; - } - - @Override - public synchronized void onActivityDefUpdate(ActivityDef activityDef) { -// initOrUpdateRateLimiters(activityDef); - } - - public synchronized void initOrUpdateRateLimiters(ActivityDef activityDef) { - -// cycleratePerThread = activityDef.getParams().takeBoolOrDefault("cyclerate_per_thread", false); - - activityDef.getParams().getOptionalNamedParameter("striderate") - .map(StrideRateSpec::new).ifPresent(sr -> this.onEvent(new ParamChange<>(sr))); - - activityDef.getParams().getOptionalNamedParameter("cyclerate", "targetrate", "rate") - .map(CycleRateSpec::new).ifPresent(sr -> this.onEvent(new ParamChange<>(sr))); - - } - - public void createOrUpdateStrideLimiter(SimRateSpec spec) { - strideLimiterSource = ThreadLocalRateLimiters.createOrUpdate(this, strideLimiterSource, spec); - } - - public void createOrUpdateCycleLimiter(SimRateSpec spec) { - cycleLimiterSource = ThreadLocalRateLimiters.createOrUpdate(this, cycleLimiterSource, spec); - } - - - /** - * Modify the provided ActivityDef with defaults for stride and cycles, if they haven't been provided, based on the - * length of the sequence as determined by the provided ratios. Also, modify the ActivityDef with reasonable - * defaults when requested. - * - * @param seq - * - The {@link OpSequence} to derive the defaults from - */ - public synchronized void setDefaultsFromOpSequence(OpSequence seq) { - Optional strideOpt = getParams().getOptionalString("stride"); - if (strideOpt.isEmpty()) { - String stride = String.valueOf(seq.getSequence().length); - logger.info(() -> "defaulting stride to " + stride + " (the sequence length)"); -// getParams().set("stride", stride); - getParams().setSilently("stride", stride); - } - - // CYCLES - Optional cyclesOpt = getParams().getOptionalString("cycles"); - if (cyclesOpt.isEmpty()) { - String cycles = getParams().getOptionalString("stride").orElseThrow(); - logger.info(() -> "defaulting cycles to " + cycles + " (the stride length)"); - this.getActivityDef().setCycles(getParams().getOptionalString("stride").orElseThrow()); - } else { - if (0 == activityDef.getCycleCount()) { - throw new RuntimeException( - "You specified cycles, but the range specified means zero cycles: " + getParams().get("cycles") - ); - } - long stride = getParams().getOptionalLong("stride").orElseThrow(); - long cycles = this.activityDef.getCycleCount(); - if (cycles < stride) { - throw new RuntimeException( - "The specified cycles (" + cycles + ") are less than the stride (" + stride + "). This means there aren't enough cycles to cause a stride to be executed." + - " If this was intended, then set stride low enough to allow it." - ); - } - } - - long cycleCount = this.activityDef.getCycleCount(); - long stride = this.activityDef.getParams().getOptionalLong("stride").orElseThrow(); - - if (0 < stride && 0 != cycleCount % stride) { - logger.warn(() -> "The stride does not evenly divide cycles. Only full strides will be executed," + - "leaving some cycles unused. (stride=" + stride + ", cycles=" + cycleCount + ')'); - } - - Optional threadSpec = activityDef.getParams().getOptionalString("threads"); - if (threadSpec.isPresent()) { - String spec = threadSpec.get(); - int processors = Runtime.getRuntime().availableProcessors(); - if ("auto".equalsIgnoreCase(spec)) { - int threads = processors * 10; - if (threads > activityDef.getCycleCount()) { - threads = (int) activityDef.getCycleCount(); - logger.info("setting threads to {} (auto) [10xCORES, cycle count limited]", threads); - } else { - logger.info("setting threads to {} (auto) [10xCORES]", threads); - } -// activityDef.setThreads(threads); - activityDef.getParams().setSilently("threads", threads); - } else if (spec.toLowerCase().matches("\\d+x")) { - String multiplier = spec.substring(0, spec.length() - 1); - int threads = processors * Integer.parseInt(multiplier); - logger.info(() -> "setting threads to " + threads + " (" + multiplier + "x)"); -// activityDef.setThreads(threads); - activityDef.getParams().setSilently("threads", threads); - } else if (spec.toLowerCase().matches("\\d+")) { - logger.info(() -> "setting threads to " + spec + " (direct)"); -// activityDef.setThreads(Integer.parseInt(spec)); - activityDef.getParams().setSilently("threads", Integer.parseInt(spec)); - } - - if (activityDef.getThreads() > activityDef.getCycleCount()) { - logger.warn(() -> "threads=" + activityDef.getThreads() + " and cycles=" + activityDef.getCycleSummary() - + ", you should have more cycles than threads."); - } - - } else if (1000 < cycleCount) { - logger.warn(() -> "For testing at scale, it is highly recommended that you " + - "set threads to a value higher than the default of 1." + - " hint: you can use threads=auto for reasonable default, or" + - " consult the topic on threads with `help threads` for" + - " more information."); - } - - if (0 < this.activityDef.getCycleCount() && seq.getOps().isEmpty()) { - throw new BasicError("You have configured a zero-length sequence and non-zero cycles. It is not possible to continue with this activity."); - } - } - - - protected OpSequence>> createOpSourceFromParsedOps( -// Map> adapterCache, -// Map> mapperCache, - List, Space>> adapters, - List pops - ) { - try { - - List ratios = new ArrayList<>(pops.size()); - - for (ParsedOp pop : pops) { - long ratio = pop.takeStaticConfigOr("ratio", 1); - ratios.add(ratio); - } - - SequencerType sequencerType = getParams() - .getOptionalString("seq") - .map(SequencerType::valueOf) - .orElse(SequencerType.bucket); - SequencePlanner>> planner = new SequencePlanner<>(sequencerType); - - for (int i = 0; i < pops.size(); i++) { - long ratio = ratios.get(i); - ParsedOp pop = pops.get(i); - - try { - if (0 == ratio) { - logger.info(() -> "skipped mapping op '" + pop.getName() + '\''); - continue; - } - - DriverAdapter, Space> adapter = adapters.get(i); - OpMapper, Space> opMapper = adapter.getOpMapper(); - LongFunction spaceFunc = adapter.getSpaceFunc(pop); - OpDispenser> dispenser = opMapper.apply(this, pop, spaceFunc); - String dryrunSpec = pop.takeStaticConfigOr("dryrun", "none"); - Dryrun dryrun = pop.takeEnumFromFieldOr(Dryrun.class, Dryrun.none, "dryrun"); - dispenser = OpFunctionComposition.wrapOptionally(adapter, dispenser, pop, dryrun); - -// if (strict) { -// optemplate.assertConsumed(); -// } - planner.addOp((OpDispenser>) dispenser, ratio); - } catch (Exception e) { - throw new OpConfigError("Error while mapping op from template named '" + pop.getName() + "': " + e.getMessage(), e); - } - } - - - return planner.resolve(); - - } catch (Exception e) { - if (e instanceof OpConfigError oce) { - throw oce; - } else { - throw new OpConfigError(e.getMessage(), workloadSource, e); - } - } - - - } - - protected List loadOpTemplates(DriverAdapter defaultDriverAdapter) { - - String tagfilter = activityDef.getParams().getOptionalString("tags").orElse(""); - - OpsDocList opsDocList = loadStmtsDocList(); - - List unfilteredOps = opsDocList.getOps(false); - List filteredOps = opsDocList.getOps(tagfilter, true); - - if (filteredOps.isEmpty()) { - // There were no ops, and it *wasn't* because they were all filtered out. - // In this case, let's try to synthesize the ops as long as at least a default driver was provided - // But if there were no ops, and there was no default driver provided, we can't continue - // There were no ops, and it was because they were all filtered out - if (!unfilteredOps.isEmpty()) { - String message = "There were no active op templates with tag filter '"+ tagfilter + "', since all " + - unfilteredOps.size() + " were filtered out. Examine the session log for details"; - NBAdvisorOutput.test(message); - //throw new BasicError(message); - } - if (defaultDriverAdapter instanceof SyntheticOpTemplateProvider sotp) { - filteredOps = sotp.getSyntheticOpTemplates(opsDocList, this.activityDef.getParams()); - Objects.requireNonNull(filteredOps); - if (filteredOps.isEmpty()) { - throw new BasicError("Attempted to create synthetic ops from driver '" + defaultDriverAdapter.getAdapterName() + '\'' + - " but no ops were created. You must provide either a workload or an op parameter. Activities require op templates."); - } - } else { - throw new BasicError(""" - No op templates were provided. You must provide one of these activity parameters: - 1) workload=some.yaml - 2) op='inline template' - 3) driver=stdout (or any other drive that can synthesize ops)"""); - } - } -// if (filteredOps.isEmpty()) { -// throw new BasicError("There were no active op templates with tag filter '" + tagfilter + '\''); -// } - -// if (filteredOps.isEmpty()) { -// throw new OpConfigError("No op templates found. You must provide either workload=... or op=..., or use " + -// "a default driver (driver=___). This includes " + -// ServiceLoader.load(DriverAdapter.class).stream() -// .filter(p -> { -// AnnotatedType[] annotatedInterfaces = p.type().getAnnotatedInterfaces(); -// for (AnnotatedType ai : annotatedInterfaces) { -// if (ai.getType().equals(SyntheticOpTemplateProvider.class)) { -// return true; -// } -// } -// return false; -// }) -// .map(d -> d.get().getAdapterName()) -// .collect(Collectors.joining(","))); -// } -// - return filteredOps; - } - - /** - * Given a function that can create an op of type from an OpTemplate, generate - * an indexed sequence of ready to call operations. - *

- * This method uses the following conventions to derive the sequence: - * - *

    - *
  1. If an 'op', 'stmt', or 'statement' parameter is provided, then it's value is - * taken as the only provided statement.
    2. - *
  2. If a 'yaml, or 'workload' parameter is provided, then the statements in that file - * are taken with their ratios
    3. - *
  3. Any provided tags filter is used to select only the op templates which have matching - * tags. If no tags are provided, then all the found op templates are included.
    4. - *
  4. The ratios and the 'seq' parameter are used to build a sequence of the ready operations, - * where the sequence length is the sum of the ratios.
    5. - *
- * - * @param - * A holder for an executable operation for the native driver used by this activity. - * @param opinit - * A function to map an OpTemplate to the executable operation form required by - * the native driver for this activity. - * @param defaultAdapter - * The adapter which will be used for any op templates with no explicit adapter - * @return The sequence of operations as determined by filtering and ratios - */ - @Deprecated(forRemoval = true) - protected OpSequence> createOpSequence(Function> opinit, boolean strict, DriverAdapter defaultAdapter) { - - List stmts = loadOpTemplates(defaultAdapter); - - List ratios = new ArrayList<>(stmts.size()); - - for (OpTemplate opTemplate : stmts) { - long ratio = opTemplate.removeParamOrDefault("ratio", 1); - ratios.add(ratio); - } - - SequencerType sequencerType = getParams() - .getOptionalString("seq") - .map(SequencerType::valueOf) - .orElse(SequencerType.bucket); - SequencePlanner> planner = new SequencePlanner<>(sequencerType); - - try { - for (int i = 0; i < stmts.size(); i++) { - long ratio = ratios.get(i); - OpTemplate optemplate = stmts.get(i); - OpDispenser driverSpecificReadyOp = opinit.apply(optemplate); - if (strict) { - optemplate.assertConsumed(); - } - planner.addOp(driverSpecificReadyOp, ratio); - } - } catch (Exception e) { - throw new OpConfigError(e.getMessage(), workloadSource, e); - } - - return planner.resolve(); - } - - protected OpsDocList loadStmtsDocList() { - - try { - String op = activityDef.getParams().getOptionalString("op").orElse(null); - String stmt = activityDef.getParams().getOptionalString("stmt", "statement").orElse(null); - String workload = activityDef.getParams().getOptionalString("workload").orElse(null); - - if ((op != null ? 1 : 0) + (stmt != null ? 1 : 0) + (workload != null ? 1 : 0) > 1) { - throw new OpConfigError("Only op, statement, or workload may be provided, not more than one."); - } - logger.debug("loadStmtsDocList #1"); - if (workload != null && OpsLoader.isJson(workload)) { - workloadSource = "commandline: (workload/json):" + workload; - return OpsLoader.loadString(workload, OpTemplateFormat.json, activityDef.getParams(), null); - } else if (workload != null && OpsLoader.isYaml(workload)) { - workloadSource = "commandline: (workload/yaml):" + workload; - return OpsLoader.loadString(workload, OpTemplateFormat.yaml, activityDef.getParams(), null); - } else if (workload != null) { - return OpsLoader.loadPath(workload, activityDef.getParams(), "activities"); - } - - logger.debug("loadStmtsDocList #2"); - if (stmt != null) { - workloadSource = "commandline: (stmt/inline): '" + stmt + "'"; - return OpsLoader.loadString(stmt, OpTemplateFormat.inline, activityDef.getParams(), null); - } - - logger.debug("loadStmtsDocList #3"); - if (op != null && OpsLoader.isJson(op)) { - workloadSource = "commandline: (op/json): '" + op + "'"; - return OpsLoader.loadString(op, OpTemplateFormat.json, activityDef.getParams(), null); - } - else if (op != null) { - workloadSource = "commandline: (op/inline): '" + op + "'"; - return OpsLoader.loadString(op, OpTemplateFormat.inline, activityDef.getParams(), null); - } - return OpsDocList.none(); - - } catch (Exception e) { - throw new OpConfigError("Error loading op templates: " + e, workloadSource, e); - } - } - - @Override - public synchronized ProgressMeterDisplay getProgressMeter() { - if (null == this.progressMeter) { - this.progressMeter = new ActivityMetricProgressMeter(this); - } - return this.progressMeter; - } - - /** - * Activities with retryable operations (when specified with the retry error handler for some - * types of error), should allow the user to specify how many retries are allowed before - * giving up on the operation. - * - * @return The number of allowable retries - */ - @Override - public int getMaxTries() { - return this.activityDef.getParams().getOptionalInteger("maxtries").orElse(10); - } - - @Override - public RunStateTally getRunStateTally() { - return tally; - } - - - @Override - public Map asResult() { - return Map.of("activity",this.getAlias()); - } - -// private final ThreadLocal cycleLimiterThreadLocal = ThreadLocal.withInitial(() -> { -// RateLimiters.createOrUpdate(this,null,new SimRateSpec() -// if (cycleratePerThread) { -// return RateLimiters.createOrUpdate(new NBThreadComponent(this),null,) -// } else { -// RateLimiters.createOrUpdate(new NBThreadComponent(this),null,activityDef) -// } -// if (getCycleLimiter() != null) { -// return RateLimiters.createOrUpdate( -// new NBThreadComponent(this), -// getCycleLimiter(), -// getCycleLimiter().getSpec()); -// } else { -// return null; -// } -// }); - -} diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/marker/ContiguousOutputChunker.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/marker/ContiguousOutputChunker.java index 3d91e89fa..5fc9d4019 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/marker/ContiguousOutputChunker.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/marker/ContiguousOutputChunker.java @@ -73,7 +73,7 @@ public class ContiguousOutputChunker implements Output { public ContiguousOutputChunker(Activity activity) { - if (!(activity.getInputDispenserDelegate().getInput(0).isContiguous())) { + if (!(activity.getInputDispenser().getInput(0).isContiguous())) { throw new RuntimeException("This type of output may not be used with non-contiguous inputs yet."); // If you are looking at this code, it's because we count updates to extents to provide // efficient marker extent handling. The ability to use segmented inputs with markers will diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/motor/CoreMotor.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/motor/CoreMotor.java index 5b150be1b..d33daa782 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/motor/CoreMotor.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/motor/CoreMotor.java @@ -323,11 +323,7 @@ public class CoreMotor implements ActivityDefObserver, Motor, Stoppable { @Override public void onActivityDefUpdate(ActivityDef activityDef) { - for (Object component : (new Object[]{input, opTracker, action, output})) { - if (component instanceof ActivityDefObserver) { - ((ActivityDefObserver) component).onActivityDefUpdate(activityDef); - } - } + ActivityDefObserver.apply(activityDef, input, opTracker, action, output); this.stride = activityDef.getParams().getOptionalInteger("stride").orElse(1); strideRateLimiter = activity.getStrideLimiter(); diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActionDispenser.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActionDispenser.java index 72ff9f1f5..f38496b06 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActionDispenser.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActionDispenser.java @@ -21,14 +21,14 @@ import io.nosqlbench.engine.api.activityapi.core.Activity; import io.nosqlbench.engine.api.activityimpl.uniform.actions.StandardAction; public class StandardActionDispenser implements ActionDispenser { - private final StandardActivity activity; + private final Activity activity; - public StandardActionDispenser(StandardActivity activity) { + public StandardActionDispenser(Activity activity) { this.activity = activity; } @Override - public StandardAction getAction(int slot) { - return new StandardAction<>(activity,slot); + public StandardAction getAction(int slot) { + return new StandardAction(activity, slot); } } diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java deleted file mode 100644 index 36619aae1..000000000 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java +++ /dev/null @@ -1,294 +0,0 @@ -/* - * Copyright (c) 2022-2023 nosqlbench - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package io.nosqlbench.engine.api.activityimpl.uniform; - -import io.nosqlbench.adapter.diag.DriverAdapterLoader; -import io.nosqlbench.adapters.api.activityconfig.OpsLoader; -import io.nosqlbench.adapters.api.activityconfig.yaml.OpTemplate; -import io.nosqlbench.adapters.api.activityconfig.yaml.OpsDocList; -import io.nosqlbench.adapters.api.activityimpl.OpDispenser; -import io.nosqlbench.adapters.api.activityimpl.OpMapper; -import io.nosqlbench.adapters.api.activityimpl.uniform.DriverAdapter; -import io.nosqlbench.adapters.api.activityimpl.uniform.Space; -import io.nosqlbench.adapters.api.activityimpl.uniform.decorators.SyntheticOpTemplateProvider; -import io.nosqlbench.adapters.api.activityimpl.uniform.flowtypes.CycleOp; -import io.nosqlbench.adapters.api.templating.ParsedOp; -import io.nosqlbench.nb.api.advisor.NBAdvisorPoint; -import io.nosqlbench.nb.api.advisor.conditions.Conditions; -import io.nosqlbench.nb.api.engine.metrics.instruments.MetricCategory; -import io.nosqlbench.nb.api.lifecycle.Shutdownable; -import io.nosqlbench.nb.api.components.core.NBComponent; -import io.nosqlbench.nb.api.config.standard.*; -import io.nosqlbench.nb.api.engine.activityimpl.ActivityDef; -import io.nosqlbench.nb.api.errors.OpConfigError; -import io.nosqlbench.nb.api.labels.NBLabels; -import io.nosqlbench.nb.api.components.events.NBEvent; -import io.nosqlbench.nb.api.components.events.ParamChange; -import io.nosqlbench.nb.api.components.events.SetThreads; -import io.nosqlbench.engine.api.activityapi.core.ActivityDefObserver; -import io.nosqlbench.engine.api.activityapi.planning.OpSequence; -import io.nosqlbench.engine.api.activityapi.simrate.CycleRateSpec; -import io.nosqlbench.engine.api.activityapi.simrate.StrideRateSpec; -import io.nosqlbench.engine.api.activityimpl.SimpleActivity; -import io.nosqlbench.nb.annotations.ServiceSelector; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; - -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; - -/** - * This is a typed activity which is expected to become the standard - * core of all new activity types. Extant NB drivers should also migrate - * to this when possible. - * - * @param - * A type of runnable which wraps the operations for this type of driver. - * @param - * The context type for the activity, AKA the 'space' for a named driver instance and its associated object graph - */ -public class StandardActivity extends SimpleActivity implements SyntheticOpTemplateProvider, ActivityDefObserver { - private static final Logger logger = LogManager.getLogger("ACTIVITY"); - private final OpSequence>> sequence; - private final ConcurrentHashMap,Space>> adapters = new ConcurrentHashMap<>(); - - public StandardActivity(NBComponent parent, ActivityDef activityDef) { - super(parent, activityDef); - NBAdvisorPoint paramsAdvisor = create().advisor(b -> b.name("Workload")); - paramsAdvisor.add(Conditions.ValidNameError); - List, Space>> adapterlist = new ArrayList<>(); - List pops = new ArrayList<>(); - ConcurrentHashMap, ? extends Space>> mappers = new ConcurrentHashMap<>(); - NBConfigModel activityModel = activityDef.getConfigModel(); - NBConfigModel yamlmodel; - NBConfigModel adapterModel; - String defaultDriverName = activityDef.getActivityDriver(); - DriverAdapter,Space> defaultAdapter = getDriverAdapter(defaultDriverName); - DriverAdapter,Space> adapter; - OpsDocList workload; - Optional yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload"); - if (yaml_loc.isPresent()) { - Map disposable = new LinkedHashMap<>(activityDef.getParams()); - workload = loadStmtsDocList(); - yamlmodel = workload.getConfigModel(); - } else { - yamlmodel = ConfigModel.of(StandardActivity.class).asReadOnly(); - } - yamlmodel.log(); - NBConfigModel supersetConfig = ConfigModel.of(StandardActivity.class).add(yamlmodel); - //NBConfigModel supersetConfig = ConfigModel.of(StandardActivity.class).add(activityModel); - // Load the op templates - List opTemplates = loadOpTemplates(defaultAdapter); - NBConfigModel combinedAdapterModel = ConfigModel.of(StandardActivity.class); - for (OpTemplate ot : opTemplates) { - logger.debug(() -> "StandardActivity.opTemplate = "+ot); - String driverName = ot.getOptionalStringParam("driver", String.class) - .or(() -> ot.getOptionalStringParam("type", String.class)) - .orElse(defaultDriverName); - if (!adapters.containsKey(driverName)) { - adapter = defaultDriverName.equals(driverName) ? defaultAdapter : getDriverAdapter(driverName); - NBConfigModel combinedModel = yamlmodel; - //NBConfigModel combinedModel = activityModel; - NBConfiguration combinedConfig = combinedModel.matchConfig(activityDef.getParams()); - if (adapter instanceof NBConfigurable configurable) { - adapterModel = configurable.getConfigModel(); - combinedAdapterModel.add(adapterModel); - supersetConfig.add(adapterModel); - combinedModel = adapterModel.add(yamlmodel); - //combinedModel = adapterModel.add(activityModel); - combinedConfig = combinedModel.matchConfig(activityDef.getParams()); - configurable.applyConfig(combinedConfig); - } - adapters.put(driverName, adapter); - mappers.put(driverName, adapter.getOpMapper()); - } else { - adapter = adapters.get(driverName); - } -// if (adapter instanceof NBConfigurable configurable) { -// adapterModel = configurable.getConfigModel(); -// adapterModel.assertValidConfig(ot.getParams()); -// } - paramsAdvisor.validateAll(ot.getParams().keySet()); - paramsAdvisor.validateAll(ot.getTags().keySet()); - paramsAdvisor.validateAll(ot.getBindings().keySet()); - adapterlist.add(adapter); - ParsedOp pop = new ParsedOp(ot, adapter.getConfiguration(), List.of(adapter.getPreprocessor()), this); - logger.debug("StandardActivity.pop="+pop); - Optional discard = pop.takeOptionalStaticValue("driver", String.class); - pops.add(pop); - } - logger.debug(() -> "StandardActivity.opTemplate loop complete"); - - paramsAdvisor.setName("Workload", "Check parameters, template, and binding names") - .logName().evaluate(); - - paramsAdvisor.clear().setName("Superset", "Check overall parameters"); - paramsAdvisor.validateAll(supersetConfig.getNamedParams().keySet()); - paramsAdvisor.logName().evaluate(); - - combinedAdapterModel.assertNoConflicts(yamlmodel.getNamedParams(), "Template"); - combinedAdapterModel.log(); - supersetConfig.assertValidConfig(activityDef.getParams().getStringStringMap()); - supersetConfig.log(); - - if (0 == mappers.keySet().stream().filter(n -> n.equals(defaultDriverName)).count()) { - logger.warn(() -> "All op templates used a different driver than the default '" + defaultDriverName + "'"); - } - - try { - sequence = createOpSourceFromParsedOps(adapterlist, pops); - } catch (Exception e) { - if (e instanceof OpConfigError) { - throw e; - } - throw new OpConfigError("Error mapping workload template to operations: " + e.getMessage(), null, e); - } - - create().gauge( - "ops_pending", - () -> this.getProgressMeter().getSummary().pending(), - MetricCategory.Core, - "The current number of operations which have not been dispatched for processing yet." - ); - create().gauge( - "ops_active", - () -> this.getProgressMeter().getSummary().current(), - MetricCategory.Core, - "The current number of operations which have been dispatched for processing, but which have not yet completed." - ); - create().gauge( - "ops_complete", - () -> this.getProgressMeter().getSummary().complete(), - MetricCategory.Core, - "The current number of operations which have been completed" - ); - } - - private DriverAdapter, Space> getDriverAdapter(String driverName) { - return Optional.of(driverName) - .flatMap(name -> ServiceSelector.of(name, ServiceLoader.load(DriverAdapterLoader.class)).get()) - .map(l -> l.load(this, NBLabels.forKV()) - ) - .orElseThrow(() -> new OpConfigError("Unable to load '" + driverName + "' driver adapter.\n"+ - "If this is a valid driver then you may need to rebuild NoSqlBench to include this driver adapter. "+ - "Change 'false' for the driver in "+ - "'./nb-adapters/pom.xml' and './nb-adapters/nb-adapters-included/pom.xml'.")); - } - - @Override - public void initActivity() { - super.initActivity(); - setDefaultsFromOpSequence(sequence); - } - - - public OpSequence>> getOpSequence() { - return sequence; - } - -// /** -// * When an adapter needs to identify an error uniquely for the purposes of -// * routing it to the correct error handler, or naming it in logs, or naming -// * metrics, override this method in your activity. -// * -// * @return A function that can reliably and safely map an instance of Throwable to a stable name. -// */ -// @Override -// public final Function getErrorNameMapper() { -// return adapter.getErrorNameMapper(); -// } - - @Override - public synchronized void onActivityDefUpdate(ActivityDef activityDef) { - super.onActivityDefUpdate(activityDef); - for (DriverAdapter adapter : adapters.values()) { - if (adapter instanceof NBReconfigurable configurable) { - NBConfigModel cfgModel = configurable.getReconfigModel(); - NBConfiguration cfg = cfgModel.matchConfig(activityDef.getParams()); - NBReconfigurable.applyMatching(cfg, List.of(configurable)); - } - } - } - -// @Override -// public synchronized void onActivityDefUpdate(final ActivityDef activityDef) { -// super.onActivityDefUpdate(activityDef); -// -// for (final DriverAdapter adapter : this.adapters.values()) -// if (adapter instanceof NBReconfigurable reconfigurable) { -// NBConfigModel cfgModel = reconfigurable.getReconfigModel(); -// final Optional op_yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload"); -// if (op_yaml_loc.isPresent()) { -// final Map disposable = new LinkedHashMap<>(activityDef.getParams()); -// final OpsDocList workload = OpsLoader.loadPath(op_yaml_loc.get(), disposable, "activities"); -// cfgModel = cfgModel.add(workload.getConfigModel()); -// } -// final NBConfiguration cfg = cfgModel.apply(activityDef.getParams()); -// reconfigurable.applyReconfig(cfg); -// } -// -// } - - @Override - public List getSyntheticOpTemplates(OpsDocList opsDocList, Map cfg) { - List opTemplates = new ArrayList<>(); - for (DriverAdapter adapter : adapters.values()) { - if (adapter instanceof SyntheticOpTemplateProvider sotp) { - List newTemplates = sotp.getSyntheticOpTemplates(opsDocList, cfg); - opTemplates.addAll(newTemplates); - } - } - return opTemplates; - } - - /** - * This is done here since driver adapters are intended to keep all of their state within - * dedicated state space types. Any space which implements {@link Shutdownable} - * will be closed when this activity shuts down. - */ - @Override - public void shutdownActivity() { - for (Map.Entry,Space>> entry : adapters.entrySet()) { - String adapterName = entry.getKey(); - DriverAdapter adapter = entry.getValue(); - if (adapter instanceof AutoCloseable autoCloseable) { - adapter.close(); - } - } - } - - @Override - public NBLabels getLabels() { - return super.getLabels(); - } - - @Override - public void onEvent(NBEvent event) { - switch (event) { - case ParamChange pc -> { - switch (pc.value()) { - case SetThreads st -> activityDef.setThreads(st.threads); - case CycleRateSpec crs -> createOrUpdateCycleLimiter(crs); - case StrideRateSpec srs -> createOrUpdateStrideLimiter(srs); - default -> super.onEvent(event); - } - } - default -> super.onEvent(event); - } - } - -} diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivityType.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivityType.java deleted file mode 100644 index 861208831..000000000 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivityType.java +++ /dev/null @@ -1,172 +0,0 @@ -/* - * Copyright (c) 2022-2023 nosqlbench - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package io.nosqlbench.engine.api.activityimpl.uniform; - - -import io.nosqlbench.adapters.api.activityimpl.uniform.DriverAdapter; -import io.nosqlbench.nb.api.components.core.NBComponent; -import io.nosqlbench.nb.api.engine.activityimpl.ActivityDef; -import io.nosqlbench.engine.api.activityapi.core.Activity; -import io.nosqlbench.engine.api.activityapi.core.ActivitiesAware; -import io.nosqlbench.engine.api.activityapi.core.ActionDispenser; -import io.nosqlbench.engine.api.activityapi.core.MotorDispenser; -import io.nosqlbench.engine.api.activityapi.input.InputDispenser; -import io.nosqlbench.engine.api.activityapi.output.OutputDispenser; -import io.nosqlbench.engine.api.activityimpl.CoreServices; -import io.nosqlbench.engine.api.activityimpl.SimpleActivity; -import io.nosqlbench.engine.api.activityimpl.action.CoreActionDispenser; -import io.nosqlbench.engine.api.activityimpl.motor.CoreMotorDispenser; - -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; - -import java.util.HashMap; -import java.util.Map; -import java.util.Optional; - -public class StandardActivityType> { - - private static final Logger logger = LogManager.getLogger("ACTIVITY"); - private final Map adapters = new HashMap<>(); - private final NBComponent parent; -// private final DriverAdapter adapter; - private final ActivityDef activityDef; - - public StandardActivityType(final DriverAdapter adapter, final ActivityDef activityDef, final NBComponent parent) { - this.parent = parent; -// this.adapter = adapter; - this.activityDef = activityDef; -// super(parent,activityDef -// .deprecate("type","driver") -// .deprecate("yaml", "workload") -// ); - adapters.put(adapter.getAdapterName(),adapter); - if (adapter instanceof ActivityDefAware) ((ActivityDefAware) adapter).setActivityDef(activityDef); - } - - public StandardActivityType(final ActivityDef activityDef, final NBComponent parent) { - this.parent = parent; - this.activityDef = activityDef; - -// super(parent,activityDef); - } - - /** - * Create an instance of an activity from the activity type. - * - * @param activityDef the definition that initializes and controls the activity. - * @return a distinct Activity instance for each call - */ - @SuppressWarnings("unchecked") - public A getActivity(final ActivityDef activityDef, final NBComponent parent) { - if (activityDef.getParams().getOptionalString("async").isPresent()) - throw new RuntimeException("This driver does not support async mode yet."); - - return (A) new StandardActivity(parent, activityDef); - } - - /** - * This method will be called once per action instance. - * - * @param activity The activity instance that will parameterize the returned ActionDispenser instance. - * @return an instance of ActionDispenser - */ - public ActionDispenser getActionDispenser(final A activity) { - return new StandardActionDispenser(activity); - } - - /** - * Create an instance of an activity that ties together all the components into a usable - * activity instance. This is the method that should be called by executor classes. - * - * @param activityDef the definition that initializez and controlls the activity. - * @param activities a map of existing activities - * @return a distinct activity instance for each call - */ - public Activity getAssembledActivity(final ActivityDef activityDef, final Map activities, final NBComponent parent) { - final A activity = this.getActivity(activityDef, parent); - - final InputDispenser inputDispenser = this.getInputDispenser(activity); - if (inputDispenser instanceof ActivitiesAware) ((ActivitiesAware) inputDispenser).setActivitiesMap(activities); - activity.setInputDispenserDelegate(inputDispenser); - - final ActionDispenser actionDispenser = this.getActionDispenser(activity); - if (actionDispenser instanceof ActivitiesAware) - ((ActivitiesAware) actionDispenser).setActivitiesMap(activities); - activity.setActionDispenserDelegate(actionDispenser); - - final OutputDispenser outputDispenser = this.getOutputDispenser(activity).orElse(null); - if ((null != outputDispenser) && (outputDispenser instanceof ActivitiesAware)) - ((ActivitiesAware) outputDispenser).setActivitiesMap(activities); - activity.setOutputDispenserDelegate(outputDispenser); - - final MotorDispenser motorDispenser = this.getMotorDispenser(activity, inputDispenser, actionDispenser, outputDispenser); - if (motorDispenser instanceof ActivitiesAware) ((ActivitiesAware) motorDispenser).setActivitiesMap(activities); - activity.setMotorDispenserDelegate(motorDispenser); - - return activity; - } - - /** - * This method will be called once per action instance. - * - * @param activity The activity instance that will parameterize the returned MarkerDispenser instance. - * @return an instance of MarkerDispenser - */ - public Optional getOutputDispenser(final A activity) { - return CoreServices.getOutputDispenser(activity); - } - - /** - * Return the InputDispenser instance that will be used by the associated activity to create Input factories - * for each thread slot. - * - * @param activity the Activity instance which will parameterize this InputDispenser - * @return the InputDispenser for the associated activity - */ - public InputDispenser getInputDispenser(final A activity) { - return CoreServices.getInputDispenser(activity); - } - - public MotorDispenser getMotorDispenser( - final A activity, - final InputDispenser inputDispenser, - final ActionDispenser actionDispenser, - final OutputDispenser outputDispenser) { - return new CoreMotorDispenser(activity, inputDispenser, actionDispenser, outputDispenser); - } - - /** - * An ActivityType can describe the canonical named types as known within that driver implementation, - * and the Java compatible types which can be assigned to them. This map is consulted when users need to select - * the name from within that driver in order to see the compatible functions which may produce a valid type for - * it. For example, a CQL users may want to know what java type (and thus what binding functions) can support - * the CQL timeuuid type. - * - * Conversely, a user may want to know what types are supported by the java.util.{@link java.util.UUID} class - * in CQL. In that case, this map will also be consulted, and multiple keys will match. - * When there are multiple answers in this way, but the driver has its own default about which one to use - * when the user's intent is ambiguous, the map should be be ordered as in {@link java.util.LinkedHashMap}, - * and the preferred form should be listed first. - * - * @return a type map indicating co-compatible associations of driver-specific type names and Java types - */ - public Map> getTypeMap() { - return Map.of(); - } - -} diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/actions/StandardAction.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/actions/StandardAction.java index a391be0cc..8ea61550c 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/actions/StandardAction.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/actions/StandardAction.java @@ -19,36 +19,29 @@ package io.nosqlbench.engine.api.activityimpl.uniform.actions; import com.codahale.metrics.Histogram; import com.codahale.metrics.Timer; import io.nosqlbench.adapters.api.activityimpl.OpDispenser; -import io.nosqlbench.adapters.api.activityimpl.uniform.BaseDriverAdapter; import io.nosqlbench.adapters.api.activityimpl.uniform.flowtypes.*; import io.nosqlbench.adapters.api.evalctx.CycleFunction; import io.nosqlbench.nb.api.engine.activityimpl.ActivityDef; import io.nosqlbench.nb.api.errors.ResultVerificationError; import io.nosqlbench.engine.api.activityapi.core.ActivityDefObserver; +import io.nosqlbench.engine.api.activityapi.core.Activity; import io.nosqlbench.engine.api.activityapi.core.SyncAction; import io.nosqlbench.engine.api.activityapi.errorhandling.modular.ErrorDetail; import io.nosqlbench.engine.api.activityapi.errorhandling.modular.NBErrorHandler; import io.nosqlbench.engine.api.activityapi.planning.OpSequence; -import io.nosqlbench.engine.api.activityimpl.uniform.StandardActivity; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import java.util.Map; import java.util.concurrent.TimeUnit; /** - This is the generified version of an Action. All driver adapters us this, as opposed + This is the unified version of an Action. All driver adapters use this, as opposed to previous NB versions where it was implemented for each driver.

This allows the API to be consolidated so that the internal machinery of NB works in a very consistent and uniform way for all users and drivers. - @param - The type of activity - @param - The type of operation */ -public class StandardAction, R extends java.util.function.LongFunction> - implements SyncAction, ActivityDefObserver -{ + */ +public class StandardAction implements SyncAction, ActivityDefObserver { private final static Logger logger = LogManager.getLogger("ACTION"); private final Timer executeTimer; private final Histogram triesHistogram; @@ -60,7 +53,7 @@ public class StandardAction, R extends java.uti private final int maxTries; private final Timer verifierTimer; - public StandardAction(A activity, int slot) { + public StandardAction(Activity activity, int slot) { this.opsequence = activity.getOpSequence(); this.maxTries = activity.getMaxTries(); bindTimer = activity.getInstrumentation().getOrCreateBindTimer(); @@ -164,6 +157,7 @@ public class StandardAction, R extends java.uti @Override public void onActivityDefUpdate(ActivityDef activityDef) { + // Activity handles rate limiter and error handler refresh internally. } } diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityExecutor.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityExecutor.java index f70631b20..2b133c335 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityExecutor.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityExecutor.java @@ -216,11 +216,7 @@ public class ActivityExecutor implements NBLabeledElement, ParameterMap.Listener if (activity.getRunState() == RunState.Running) { adjustMotorCountToThreadParam(activity.getActivityDef()); } - motors.stream() - .filter(m -> (m instanceof ActivityDefObserver)) -// .filter(m -> m.getSlotStateTracker().getSlotState() != RunState.Uninitialized) -// .filter(m -> m.getSlotStateTracker().getSlotState() != RunState.Starting) - .forEach(m -> ((ActivityDefObserver) m).onActivityDefUpdate(activityDef)); + ActivityDefObserver.apply(activityDef, motors.toArray()); } } @@ -260,7 +256,7 @@ public class ActivityExecutor implements NBLabeledElement, ParameterMap.Listener // Create motor slots try { while (motors.size() < activityDef.getThreads()) { - Motor motor = activity.getMotorDispenserDelegate().getMotor(activityDef, motors.size()); + Motor motor = activity.getMotorDispenser().getMotor(activityDef, motors.size()); logger.trace(() -> "Starting cycle motor thread:" + motor); motors.add(motor); } @@ -612,5 +608,3 @@ public class ActivityExecutor implements NBLabeledElement, ParameterMap.Listener } } } - - diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityLoader.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityLoader.java index 4246ef46c..c0c40dd40 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityLoader.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityLoader.java @@ -19,7 +19,14 @@ package io.nosqlbench.engine.core.lifecycle.activity; import io.nosqlbench.nb.api.engine.activityimpl.ActivityDef; import io.nosqlbench.nb.api.components.core.NBComponent; import io.nosqlbench.engine.api.activityapi.core.Activity; -import io.nosqlbench.engine.api.activityimpl.uniform.StandardActivityType; +import io.nosqlbench.engine.api.activityapi.core.ActivitiesAware; +import io.nosqlbench.engine.api.activityapi.core.ActionDispenser; +import io.nosqlbench.engine.api.activityapi.core.MotorDispenser; +import io.nosqlbench.engine.api.activityapi.input.InputDispenser; +import io.nosqlbench.engine.api.activityapi.output.OutputDispenser; +import io.nosqlbench.engine.api.activityimpl.CoreServices; +import io.nosqlbench.engine.api.activityimpl.motor.CoreMotorDispenser; +import io.nosqlbench.engine.api.activityimpl.uniform.StandardActionDispenser; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -39,9 +46,37 @@ public class ActivityLoader { } public synchronized Activity loadActivity(ActivityDef activityDef, final NBComponent parent) { - activityDef= activityDef.deprecate("yaml","workload").deprecate("type","driver"); - final Activity activity = new StandardActivityType<>(activityDef, parent).getAssembledActivity(activityDef, this.activityMap, parent); - this.activityMap.put(activity.getAlias(),activity); + activityDef = activityDef.deprecate("yaml", "workload").deprecate("type", "driver"); + + // Create the activity instance directly (using Activity for compatibility) + final Activity activity = new Activity(parent, activityDef); + + // Assemble the dispensers like StandardActivityType did + final InputDispenser inputDispenser = CoreServices.getInputDispenser(activity); + if (inputDispenser instanceof ActivitiesAware) { + ((ActivitiesAware) inputDispenser).setActivitiesMap(activityMap); + } + activity.setInputDispenser(inputDispenser); + + final ActionDispenser actionDispenser = new StandardActionDispenser(activity); + if (actionDispenser instanceof ActivitiesAware) { + ((ActivitiesAware) actionDispenser).setActivitiesMap(activityMap); + } + activity.setActionDispenser(actionDispenser); + + final OutputDispenser outputDispenser = CoreServices.getOutputDispenser(activity).orElse(null); + if ((null != outputDispenser) && (outputDispenser instanceof ActivitiesAware)) { + ((ActivitiesAware) outputDispenser).setActivitiesMap(activityMap); + } + activity.setOutputDispenser(outputDispenser); + + final MotorDispenser motorDispenser = new CoreMotorDispenser<>(activity, inputDispenser, actionDispenser, outputDispenser); + if (motorDispenser instanceof ActivitiesAware) { + ((ActivitiesAware) motorDispenser).setActivitiesMap(activityMap); + } + activity.setMotorDispenser(motorDispenser); + + this.activityMap.put(activity.getAlias(), activity); ActivityLoader.logger.debug("Resolved activity for alias '{}'", activityDef.getAlias()); return activity; } diff --git a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityTypeLoader.java b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityTypeLoader.java index 6d0ea3721..a86dbf661 100644 --- a/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityTypeLoader.java +++ b/nb-engine/nb-engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityTypeLoader.java @@ -18,6 +18,7 @@ package io.nosqlbench.engine.core.lifecycle.activity; import io.nosqlbench.adapter.diag.DriverAdapterLoader; import io.nosqlbench.adapters.api.activityimpl.uniform.DriverAdapter; +import io.nosqlbench.engine.api.activityimpl.uniform.ActivityDefAware; import io.nosqlbench.nb.api.components.core.NBComponent; import io.nosqlbench.nb.api.nbio.Content; import io.nosqlbench.nb.api.nbio.NBIO; @@ -25,7 +26,6 @@ import io.nosqlbench.nb.api.engine.activityimpl.ActivityDef; import io.nosqlbench.nb.api.errors.BasicError; import io.nosqlbench.nb.api.spi.SimpleServiceLoader; import io.nosqlbench.nb.api.system.NBEnvironment; -import io.nosqlbench.engine.api.activityimpl.uniform.StandardActivityType; import io.nosqlbench.nb.annotations.Maturity; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -118,7 +118,7 @@ public class ActivityTypeLoader { return urlsToAdd; } - public Optional load(final ActivityDef activityDef, final NBComponent parent) { + public Optional> load(final ActivityDef activityDef, final NBComponent parent) { String driverName = activityDef.getParams() .getOptionalString("driver", "type") @@ -135,20 +135,19 @@ public class ActivityTypeLoader { }) .ifPresent(this::extendClassLoader); - return getDriverAdapter(driverName,activityDef,parent); + return getDriverAdapter(driverName, activityDef, parent); } - private Optional getDriverAdapter(final String activityTypeName, final ActivityDef activityDef, final NBComponent parent) { + private Optional> getDriverAdapter(final String activityTypeName, final ActivityDef activityDef, final NBComponent parent) { final Optional oda = this.DRIVERADAPTER_SPI_FINDER.getOptionally(activityTypeName); - if (oda.isPresent()) { - final DriverAdapter driverAdapter = oda.get(); - - final StandardActivityType activityType = new StandardActivityType<>(driverAdapter, activityDef, parent); - return Optional.of(activityType); - } - return Optional.empty(); + return oda.map(adapter -> { + if (adapter instanceof ActivityDefAware aware) { + aware.setActivityDef(activityDef); + } + return adapter; + }); } public Set getAllSelectors() { diff --git a/nb-engine/nb-engine-core/src/test/java/io/nosqlbench/engine/core/ActivityExecutorTest.java b/nb-engine/nb-engine-core/src/test/java/io/nosqlbench/engine/core/ActivityExecutorTest.java index 65766b97f..8bacf9bce 100644 --- a/nb-engine/nb-engine-core/src/test/java/io/nosqlbench/engine/core/ActivityExecutorTest.java +++ b/nb-engine/nb-engine-core/src/test/java/io/nosqlbench/engine/core/ActivityExecutorTest.java @@ -24,7 +24,6 @@ import io.nosqlbench.engine.api.activityapi.input.Input; import io.nosqlbench.engine.api.activityapi.input.InputDispenser; import io.nosqlbench.engine.api.activityapi.output.OutputDispenser; import io.nosqlbench.engine.api.activityimpl.CoreServices; -import io.nosqlbench.engine.api.activityimpl.SimpleActivity; import io.nosqlbench.engine.api.activityimpl.action.CoreActionDispenser; import io.nosqlbench.engine.api.activityimpl.input.CoreInputDispenser; import io.nosqlbench.engine.api.activityimpl.motor.CoreMotor; @@ -56,10 +55,10 @@ class ActivityExecutorTest { // OutputDispenser tdisp = CoreServices.getOutputDispenser(activity).orElse(null); // // final MotorDispenser mdisp = new CoreMotorDispenser(activity, inputDispenser, adisp, tdisp); -// activity.setActionDispenserDelegate(adisp); -// activity.setOutputDispenserDelegate(tdisp); -// activity.setInputDispenserDelegate(inputDispenser); -// activity.setMotorDispenserDelegate(mdisp); +// activity.setActionDispenser(adisp); +// activity.setOutputDispenser(tdisp); +// activity.setInputDispenser(inputDispenser); +// activity.setMotorDispenser(mdisp); // // final ExecutorService executor = Executors.newCachedThreadPool(); // ActivityExecutor activityExecutor = new ActivityExecutor(activity, "test-restart"); @@ -93,8 +92,7 @@ class ActivityExecutorTest { + "cycles=1000;initdelay=2000;" + "labels=invalid-name:valid" + "-value"); - new ActivityTypeLoader().load(activityDef, TestComponent.INSTANCE); - Activity activity = new DelayedInitActivity(activityDef); + Activity activity = new DelayedInitActivity(activityDef); fail("Expected an Advisor exception"); } catch (RuntimeException e) { assertThat(e.toString().contains("error")); @@ -106,7 +104,6 @@ class ActivityExecutorTest { synchronized void testDelayedStartSanity() { ActivityDef activityDef = ActivityDef.parseActivityDef("driver=diag;alias=test_delayed_start;cycles=1000;initdelay=2000;"); - new ActivityTypeLoader().load(activityDef, TestComponent.INSTANCE); Activity activity = new DelayedInitActivity(activityDef); final InputDispenser inputDispenser = new CoreInputDispenser(activity); @@ -114,10 +111,10 @@ class ActivityExecutorTest { final OutputDispenser outputDispenser = CoreServices.getOutputDispenser(activity).orElse(null); MotorDispenser motorDispenser = new CoreMotorDispenser(activity, inputDispenser, actionDispenser, outputDispenser); - activity.setActionDispenserDelegate(actionDispenser); - activity.setOutputDispenserDelegate(outputDispenser); - activity.setInputDispenserDelegate(inputDispenser); - activity.setMotorDispenserDelegate(motorDispenser); + activity.setActionDispenser(actionDispenser); + activity.setOutputDispenser(outputDispenser); + activity.setInputDispenser(inputDispenser); + activity.setMotorDispenser(motorDispenser); ActivityExecutor activityExecutor = new ActivityExecutor(activity); @@ -142,22 +139,22 @@ class ActivityExecutorTest { final ActivityDef activityDef = ActivityDef.parseActivityDef("driver=diag;alias=test_dynamic_params;cycles=1000;initdelay=5000;"); new ActivityTypeLoader().load(activityDef,TestComponent.INSTANCE); - Activity simpleActivity = new SimpleActivity(TestComponent.INSTANCE,activityDef); + Activity activity = new Activity(TestComponent.INSTANCE, activityDef); -// this.getActivityMotorFactory(this.motorActionDelay(999), new AtomicInput(simpleActivity,activityDef)); +// this.getActivityMotorFactory(this.motorActionDelay(999), new AtomicInput(activity,activityDef)); - final InputDispenser inputDispenser = new CoreInputDispenser(simpleActivity); - final ActionDispenser actionDispenser = new CoreActionDispenser(simpleActivity); - final OutputDispenser outputDispenser = CoreServices.getOutputDispenser(simpleActivity).orElse(null); + final InputDispenser inputDispenser = new CoreInputDispenser(activity); + final ActionDispenser actionDispenser = new CoreActionDispenser(activity); + final OutputDispenser outputDispenser = CoreServices.getOutputDispenser(activity).orElse(null); - MotorDispenser motorDispenser = new CoreMotorDispenser<>(simpleActivity, + MotorDispenser motorDispenser = new CoreMotorDispenser<>(activity, inputDispenser, actionDispenser, outputDispenser); - simpleActivity.setActionDispenserDelegate(actionDispenser); - simpleActivity.setInputDispenserDelegate(inputDispenser); - simpleActivity.setMotorDispenserDelegate(motorDispenser); + activity.setActionDispenser(actionDispenser); + activity.setInputDispenser(inputDispenser); + activity.setMotorDispenser(motorDispenser); - ActivityExecutor activityExecutor = new ActivityExecutor(simpleActivity); + ActivityExecutor activityExecutor = new ActivityExecutor(activity); activityDef.setThreads(5); ForkJoinTask executionResultForkJoinTask = ForkJoinPool.commonPool().submit(activityExecutor); @@ -192,7 +189,7 @@ class ActivityExecutorTest { return new MotorDispenser<>() { @Override public Motor getMotor(final ActivityDef activityDef, final int slotId) { - final Activity activity = new SimpleActivity(TestComponent.INSTANCE,activityDef); + final Activity activity = new Activity(TestComponent.INSTANCE, activityDef); final Motor cm = new CoreMotor<>(activity, slotId, ls); cm.setAction(lc); return cm; @@ -215,7 +212,7 @@ class ActivityExecutorTest { } - private static class DelayedInitActivity extends SimpleActivity { + private static class DelayedInitActivity extends Activity { private static final Logger logger = LogManager.getLogger(DelayedInitActivity.class); public DelayedInitActivity(final ActivityDef activityDef) { diff --git a/nb-engine/nb-engine-core/src/test/java/io/nosqlbench/engine/core/CoreMotorTest.java b/nb-engine/nb-engine-core/src/test/java/io/nosqlbench/engine/core/CoreMotorTest.java index e210ddd23..5bf3a3c23 100644 --- a/nb-engine/nb-engine-core/src/test/java/io/nosqlbench/engine/core/CoreMotorTest.java +++ b/nb-engine/nb-engine-core/src/test/java/io/nosqlbench/engine/core/CoreMotorTest.java @@ -22,7 +22,6 @@ import io.nosqlbench.engine.api.activityapi.core.Action; import io.nosqlbench.engine.api.activityapi.core.Activity; import io.nosqlbench.engine.api.activityapi.core.Motor; import io.nosqlbench.engine.api.activityapi.core.SyncAction; -import io.nosqlbench.engine.api.activityimpl.SimpleActivity; import io.nosqlbench.engine.api.activityimpl.motor.CoreMotor; import io.nosqlbench.engine.core.fortesting.BlockingSegmentInput; import org.junit.jupiter.api.Test; @@ -37,7 +36,7 @@ public class CoreMotorTest { @Test public void testBasicActivityMotor() { - final Activity activity = new SimpleActivity( + final Activity activity = new Activity( new TestComponent("testing", "coremotor"), ActivityDef.parseActivityDef("alias=foo") ); @@ -60,7 +59,7 @@ public class CoreMotorTest { @Test public void testIteratorStride() { - SimpleActivity activity = new SimpleActivity(TestComponent.INSTANCE, "stride=3"); + Activity activity = new Activity(TestComponent.INSTANCE, "stride=3"); final BlockingSegmentInput lockstepper = new BlockingSegmentInput(); final Motor cm1 = new CoreMotor(activity, 1L, lockstepper); final AtomicLongArray ary = new AtomicLongArray(10); diff --git a/nbr-demos/src/test/java/io/nosqlbench/nbr/tests/TestSessionMetrics.java b/nbr-demos/src/test/java/io/nosqlbench/nbr/tests/TestSessionMetrics.java index 20e11f68e..f9280ea8a 100644 --- a/nbr-demos/src/test/java/io/nosqlbench/nbr/tests/TestSessionMetrics.java +++ b/nbr-demos/src/test/java/io/nosqlbench/nbr/tests/TestSessionMetrics.java @@ -21,7 +21,6 @@ package io.nosqlbench.nbr.tests; import com.codahale.metrics.Gauge; import io.nosqlbench.adapters.api.activityimpl.uniform.Space; import io.nosqlbench.engine.api.activityapi.core.Activity; -import io.nosqlbench.engine.api.activityimpl.uniform.StandardActivity; import io.nosqlbench.engine.core.lifecycle.commands.CMD_await; import io.nosqlbench.engine.core.lifecycle.commands.CMD_start; import io.nosqlbench.engine.core.lifecycle.scenario.container.NBBufferedContainer; @@ -117,7 +116,7 @@ public class TestSessionMetrics { while (all.hasNext()) { NBComponent component = all.next(); - if (component instanceof StandardActivity activity) { + if (component instanceof Activity activity) { activities.add(activity); } if (component instanceof Space space) { diff --git a/nbr/src/main/java/io/nosqlbench/scenarios/simframe/optimizers/CMD_reset.java b/nbr/src/main/java/io/nosqlbench/scenarios/simframe/optimizers/CMD_reset.java index 4fa690440..a34eaf737 100644 --- a/nbr/src/main/java/io/nosqlbench/scenarios/simframe/optimizers/CMD_reset.java +++ b/nbr/src/main/java/io/nosqlbench/scenarios/simframe/optimizers/CMD_reset.java @@ -94,7 +94,7 @@ public class CMD_reset extends NBBaseCommand { //TODO: This needs to be reworked, but simply calling controller.start on the flywheel results in 2 // copies of the activity running simultaneously. This is a temporary workaround. SimFrameUtils.awaitActivity(flywheel); - flywheel.getMotorDispenserDelegate().getMotor(flywheel.getActivityDef(), 0).run(); + flywheel.getMotorDispenser().getMotor(flywheel.getActivityDef(), 0).run(); } return null;