Line data Source code
1 : import * as React from 'react';
2 :
3 : import { ConnectionHealth, connectionCachesQueryResults, type ConnectionState } from '../../connections/connection_state.js';
4 : import { connectorOfDetails } from '../../connections/connector_descriptor.js';
5 : import { useConnectionRegistry } from '../../connections/connection_registry.js';
6 : import { useCancelQuery, useQueryExecutor } from '../../connections/query_executor.js';
7 : import { useSessionScriptsDispatch, useSessionScriptsRegistry, useSessionScriptsRegistryReader, type ModifySessionScripts } from './session_scripts_registry.js';
8 : import { createExecutionMounts, isSessionExecutionRunning, runSessionExecution, subscribeSessionExecutions, type SessionExecutionState } from './session_execution.js';
9 : import { useComputeDatabase } from '../../compute/compute_connection_provider.js';
10 : import { supportsParquet } from '../../platform/hyperdb/hyperdb_parquet_export.js';
11 : import { registerScriptRunServices, type ScriptRunServices } from './script_run.js';
12 : import { computeQueryCacheKeyForConnection } from '../../connections/query_executor.js';
13 : import { useQueryResultCache } from '../../storage/query_result_cache_provider.js';
14 : import { scriptConnectionIds, SET_ATTACHMENT_COLUMNS } from './session_scripts.js';
15 : import { QueryType } from '../../query/query_execution_state.js';
16 : import { useLogger } from '../../platform/logger/logger_provider.js';
17 : import { mountSessionAttachments, readAttachmentColumns } from '../attachments/attachment_mounts.js';
18 : import { attachmentStamp } from '../attachments/session_attachments.js';
19 : import { stringifyError } from '../../platform/logger/logger.js';
20 :
21 1 : const LOG_CTX = 'session_execution';
22 :
23 : /// The options of a run of a session
24 : export interface SessionExecutionOptions {
25 : /// Run the scripts that a change of these inputs affects. A run that holds the session stops, the run starts
26 : /// once the session is free, with the inputs of every change since.
27 : inputs?: readonly string[];
28 : }
29 :
30 : interface ExecutionContext {
31 : states: Map<string, SessionExecutionState>;
32 : start: (sessionId: string, modify: ModifySessionScripts, options?: SessionExecutionOptions) => void;
33 : stop: (sessionId: string) => void;
34 : }
35 :
36 1 : const CONTEXT = React.createContext<ExecutionContext | null>(null);
37 :
38 : /// The number of statements of a session run that a connection executes at once
39 1 : function maxConcurrentQueries(connection: ConnectionState | undefined): number {
40 1 : return connection == null ? 1 : connectorOfDetails(connection.details).maxConcurrentQueries(connection.details.value);
41 1 : }
42 :
43 : /// Does a connection run a query? The SQLFrame queries of computations run on the compute database,
44 : /// the connection only tracks them.
45 1 : function runsQueries(connection: ConnectionState): boolean {
46 1 : for (const query of connection.queriesActive.values()) {
47 1 : if (query.queryMetadata.queryType !== QueryType.INTERNAL_SQLFRAME) return true;
48 : }
49 1 : return false;
50 1 : }
51 :
52 : /// The attempts to wait for a run of a session to leave its connections
53 1 : const FREE_ATTEMPTS = 50;
54 1 : const FREE_INTERVAL_MS = 100;
55 :
56 : /// Wait until a session is free for a run: no run holds it and its connections are idle. Gives up after a while, a
57 : /// connection may stay busy.
58 0 : function whenSessionFree(isFree: () => boolean): Promise<boolean> {
59 0 : return new Promise(resolve => {
60 0 : let attempts = 0;
61 0 : const check = () => {
62 0 : if (isFree()) {
63 0 : resolve(true);
64 0 : } else if (++attempts > FREE_ATTEMPTS) {
65 0 : resolve(false);
66 : } else {
67 0 : setTimeout(check, FREE_INTERVAL_MS);
68 : }
69 0 : };
70 0 : check();
71 0 : });
72 0 : }
73 :
74 1 : export function sessionExecutionIsBusy(state: SessionExecutionState | null | undefined): boolean {
75 1 : return state?.phase === 'planning' || state?.phase === 'running' || state?.phase === 'stopping';
76 1 : }
77 :
78 : /// The run of a session. `busy` while the whole session runs, `runActive` while any run holds the session, a run of
79 : /// the session or of one of its scripts: the session starts no other run then.
80 1 : export function useSessionExecution(sessionId: string) {
81 1 : const context = React.useContext(CONTEXT);
82 1 : const state = context?.states.get(sessionId) ?? null;
83 1 : const start = context?.start;
84 1 : const stop = context?.stop;
85 1 : const held = React.useSyncExternalStore(subscribeSessionExecutions, () => isSessionExecutionRunning(sessionId));
86 1 : return React.useMemo(() => {
87 1 : const busy = sessionExecutionIsBusy(state);
88 1 : return {
89 1 : state,
90 1 : busy,
91 1 : runActive: busy || held,
92 1 : start: (modify: ModifySessionScripts, options?: SessionExecutionOptions) => start?.(sessionId, modify, options),
93 1 : stop: () => stop?.(sessionId),
94 1 : };
95 1 : }, [state, held, start, stop, sessionId]);
96 1 : }
97 :
98 1 : export function SessionExecutionProvider(props: React.PropsWithChildren) {
99 1 : const [registry] = useSessionScriptsRegistry();
100 : // The registry as the last update left it, also before the next render
101 1 : const readRegistry = useSessionScriptsRegistryReader();
102 1 : const [connections] = useConnectionRegistry();
103 1 : const executeQuery = useQueryExecutor();
104 1 : const cancelQuery = useCancelQuery();
105 1 : const computeDb = useComputeDatabase();
106 1 : const queryResultCache = useQueryResultCache();
107 1 : const logger = useLogger();
108 1 : const [states, setStates] = React.useState(new Map<string, SessionExecutionState>());
109 1 : const runs = React.useRef(new Map<string, AbortController>());
110 1 : const runningDatabases = React.useRef(new Set<string>());
111 : // The input changes that wait for the session, by session id
112 1 : const pendingInputs = React.useRef(new Map<string, { modify: ModifySessionScripts; inputs: Set<string> }>());
113 :
114 1 : React.useEffect(() => () => {
115 1 : for (const controller of runs.current.values()) controller.abort();
116 1 : }, []);
117 :
118 : // The columns of the Parquet attachments, read once per file and stamp with the compute database
119 1 : const dispatchSessionScripts = useSessionScriptsDispatch();
120 1 : const describedAttachments = React.useRef(new Set<string>());
121 1 : React.useEffect(() => {
122 1 : if (!supportsParquet(computeDb)) return;
123 1 : for (const session of registry.sessionScriptsMap.values()) {
124 1 : for (const attachment of session.attachments ?? []) {
125 1 : if (attachment.columns != null) continue;
126 1 : const stamp = attachmentStamp(attachment);
127 1 : const key = `${session.sessionId}/${attachment.fileName}@${stamp}`;
128 1 : if (describedAttachments.current.has(key)) continue;
129 1 : describedAttachments.current.add(key);
130 1 : const { sessionId } = session;
131 1 : const fileName = attachment.fileName;
132 1 : readAttachmentColumns(computeDb, sessionId, fileName)
133 1 : .then(columns => dispatchSessionScripts(sessionId, { type: SET_ATTACHMENT_COLUMNS, value: { fileName, stamp, columns } }))
134 1 : .catch(error => logger.warn('Reading the columns of an attachment failed', { sessionId, fileName, error: stringifyError(error) }, LOG_CTX));
135 1 : }
136 : }
137 1 : }, [registry, computeDb, dispatchSessionScripts, logger]);
138 1 : React.useEffect(() => {
139 1 : for (const [id, controller] of runs.current) {
140 1 : const session = registry.sessionScriptsMap.get(id);
141 1 : if (!session || scriptConnectionIds(session).some(connectionId =>
142 1 : connections.connections.get(connectionId)?.connectionHealth !== ConnectionHealth.ONLINE)) {
143 1 : controller.abort();
144 : }
145 1 : }
146 1 : }, [registry, connections]);
147 :
148 : // The provider re-renders with every registry change.
149 : // Keep the context value stable and read the latest registries when an execution starts.
150 1 : const latest = React.useRef({ registry, connections, executeQuery, cancelQuery, computeDb, queryResultCache, logger });
151 1 : latest.current = { registry, connections, executeQuery, cancelQuery, computeDb, queryResultCache, logger };
152 :
153 : // A run of a script plans it with the scripts it references, the services read the latest registries
154 1 : const runServices = React.useMemo<ScriptRunServices>(() => ({
155 1 : createMounts: () => {
156 1 : const { computeDb, queryResultCache } = latest.current;
157 1 : return queryResultCache != null && supportsParquet(computeDb) ? createExecutionMounts(queryResultCache, computeDb) : null;
158 1 : },
159 1 : mountAttachments: async sessionId => {
160 0 : const { computeDb } = latest.current;
161 0 : if (!supportsParquet(computeDb)) throw new Error('Scripts that read attachments need the compute database');
162 0 : await mountSessionAttachments(computeDb, sessionId);
163 0 : },
164 1 : cacheKey: async (connectionId, signature) => {
165 1 : const connection = latest.current.connections.connections.get(connectionId);
166 1 : return connection == null ? null : await computeQueryCacheKeyForConnection(connection.details, signature);
167 1 : },
168 1 : cachesResults: connectionId => {
169 1 : const connection = latest.current.connections.connections.get(connectionId);
170 1 : return connection != null && connectionCachesQueryResults(connection);
171 1 : },
172 1 : get cache() { return latest.current.queryResultCache; },
173 1 : maxConcurrentQueries: connectionId => maxConcurrentQueries(latest.current.connections.connections.get(connectionId)),
174 1 : liveSession: sessionId => (readRegistry?.() ?? latest.current.registry).sessionScriptsMap.get(sessionId) ?? null,
175 1 : }), [readRegistry]);
176 : // A layout effect, the passive effects of the children read the services, e.g. the cache indicators of the
177 : // cards. React runs passive effects child first and a registry change unregisters the services before it
178 : // registers them again: with a passive effect the children would plan with the plain services that cache
179 : // nothing in between.
180 1 : React.useLayoutEffect(() => {
181 1 : const instances = new Set([...registry.sessionScriptsMap.values()].map(session => session.instance));
182 1 : for (const instance of instances) registerScriptRunServices(instance, runServices);
183 1 : return () => {
184 1 : for (const instance of instances) registerScriptRunServices(instance, null);
185 1 : };
186 1 : }, [registry, runServices]);
187 :
188 1 : const startRef = React.useRef<(sessionId: string, modify: ModifySessionScripts, options?: SessionExecutionOptions) => void>(() => {});
189 1 : const start = React.useCallback((sessionId: string, modifySessionScripts: ModifySessionScripts, options?: SessionExecutionOptions) => {
190 1 : const { connections, executeQuery, cancelQuery, computeDb, queryResultCache, logger } = latest.current;
191 : // The registry as the last update left it, an input change starts its run right after writing its tags
192 1 : const registry = readRegistry?.() ?? latest.current.registry;
193 1 : const session = registry.sessionScriptsMap.get(sessionId);
194 1 : if (!session || session.scriptOrder.length === 0) return;
195 : // A run takes every connection of the session, all of them must be idle
196 1 : const connectionIds = scriptConnectionIds(session);
197 1 : const isFree = () => {
198 1 : const connections = latest.current.connections;
199 1 : return !runs.current.has(sessionId) && !isSessionExecutionRunning(sessionId) && connectionIds.every(connectionId => {
200 1 : const connection = connections.connections.get(connectionId);
201 1 : return !runningDatabases.current.has(connectionId)
202 1 : && connection?.connectionHealth === ConnectionHealth.ONLINE
203 1 : && !runsQueries(connection);
204 1 : });
205 1 : };
206 1 : if (options?.inputs != null) {
207 0 : const pending = pendingInputs.current.get(sessionId);
208 0 : const inputs = new Set([...(pending?.inputs ?? []), ...options.inputs]);
209 0 : if (pending != null || !isFree()) {
210 : // The latest values win: stop the run that holds the session and start once the session is free
211 0 : pendingInputs.current.set(sessionId, { modify: modifySessionScripts, inputs });
212 0 : runs.current.get(sessionId)?.abort();
213 0 : if (pending == null) {
214 0 : void whenSessionFree(isFree).then(free => {
215 0 : const next = pendingInputs.current.get(sessionId);
216 0 : pendingInputs.current.delete(sessionId);
217 0 : if (next != null && free) startRef.current(sessionId, next.modify, { inputs: [...next.inputs] });
218 0 : });
219 : }
220 0 : return;
221 0 : }
222 0 : options = { inputs: [...inputs] };
223 0 : }
224 1 : if (!isFree()) return;
225 1 : const controller = new AbortController();
226 1 : runs.current.set(sessionId, controller);
227 1 : for (const connectionId of connectionIds) runningDatabases.current.add(connectionId);
228 1 : void runSessionExecution({
229 1 : session,
230 1 : getSession: () => (readRegistry?.() ?? latest.current.registry).sessionScriptsMap.get(sessionId) ?? null,
231 1 : inputs: options?.inputs,
232 1 : executeQuery, cancelQuery, modifySessionScripts,
233 1 : maxConcurrentQueries: connectionId => maxConcurrentQueries(connections.connections.get(connectionId)),
234 : // The embedded Hyper scans the results of the scripts of other connections, mounted from the compute database
235 1 : mounts: queryResultCache != null && supportsParquet(computeDb) ? createExecutionMounts(queryResultCache, computeDb) : null,
236 1 : mountAttachments: runServices.mountAttachments,
237 1 : resultKey: runServices.cacheKey,
238 1 : cachesResults: runServices.cachesResults,
239 1 : cache: queryResultCache,
240 1 : signal: controller.signal,
241 1 : logger,
242 1 : onState: state => setStates(previous => new Map(previous).set(sessionId, state)),
243 1 : }).finally(() => {
244 1 : runs.current.delete(sessionId);
245 1 : for (const connectionId of connectionIds) runningDatabases.current.delete(connectionId);
246 1 : });
247 1 : }, [runServices, readRegistry]);
248 1 : startRef.current = start;
249 1 : const stop = React.useCallback((id: string) => runs.current.get(id)?.abort(), []);
250 1 : const value = React.useMemo<ExecutionContext>(() => ({ states, start, stop }), [states, start, stop]);
251 :
252 1 : return <CONTEXT.Provider value={value}>
253 1 : {props.children}
254 1 : </CONTEXT.Provider>;
255 1 : }
|
