Migrate to private class properties/methods
This commit is contained in:
Jamie Kyle
GitHub
parent
7dbe57084b
commit
aa9f53df57
100 changed files with 3795 additions and 3944 deletions
217
ts/sql/main.ts
217
ts/sql/main.ts
|
|
@ -116,31 +116,25 @@ export type QueryStatsOptions = {
|
|||
};
|
||||
|
||||
export class MainSQL {
|
||||
private readonly pool = new Array<PoolEntry>();
|
||||
|
||||
private pauseWaiters: Array<() => void> | undefined;
|
||||
|
||||
private isReady = false;
|
||||
|
||||
private onReady: Promise<void> | undefined;
|
||||
|
||||
private readonly onExit: Promise<unknown>;
|
||||
readonly #pool = new Array<PoolEntry>();
|
||||
#pauseWaiters: Array<() => void> | undefined;
|
||||
#isReady = false;
|
||||
#onReady: Promise<void> | undefined;
|
||||
readonly #onExit: Promise<unknown>;
|
||||
|
||||
// Promise resolve callbacks for corruption and readonly errors.
|
||||
private errorResolvers = new Array<KnownErrorResolverType>();
|
||||
#errorResolvers = new Array<KnownErrorResolverType>();
|
||||
|
||||
private seq = 0;
|
||||
|
||||
private logger?: LoggerType;
|
||||
#seq = 0;
|
||||
#logger?: LoggerType;
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
private onResponse = new Map<number, PromisePair<any>>();
|
||||
#onResponse = new Map<number, PromisePair<any>>();
|
||||
|
||||
private shouldTimeQueries = false;
|
||||
#shouldTimeQueries = false;
|
||||
#shouldTrackQueryStats = false;
|
||||
|
||||
private shouldTrackQueryStats = false;
|
||||
|
||||
private queryStats?: {
|
||||
#queryStats?: {
|
||||
start: number;
|
||||
statsByQuery: Map<string, QueryStatsType>;
|
||||
};
|
||||
|
|
@ -148,12 +142,12 @@ export class MainSQL {
|
|||
constructor() {
|
||||
const exitPromises = new Array<Promise<void>>();
|
||||
for (let i = 0; i < WORKER_COUNT; i += 1) {
|
||||
const { worker, onExit } = this.createWorker();
|
||||
this.pool.push({ worker, load: 0 });
|
||||
const { worker, onExit } = this.#createWorker();
|
||||
this.#pool.push({ worker, load: 0 });
|
||||
|
||||
exitPromises.push(onExit);
|
||||
}
|
||||
this.onExit = Promise.all(exitPromises);
|
||||
this.#onExit = Promise.all(exitPromises);
|
||||
}
|
||||
|
||||
public async initialize({
|
||||
|
|
@ -162,19 +156,19 @@ export class MainSQL {
|
|||
key,
|
||||
logger,
|
||||
}: InitializeOptions): Promise<void> {
|
||||
if (this.isReady || this.onReady) {
|
||||
if (this.#isReady || this.#onReady) {
|
||||
throw new Error('Already initialized');
|
||||
}
|
||||
|
||||
this.shouldTimeQueries = Boolean(process.env.TIME_QUERIES);
|
||||
this.#shouldTimeQueries = Boolean(process.env.TIME_QUERIES);
|
||||
|
||||
this.logger = logger;
|
||||
this.#logger = logger;
|
||||
|
||||
this.onReady = (async () => {
|
||||
const primary = this.pool[0];
|
||||
const rest = this.pool.slice(1);
|
||||
this.#onReady = (async () => {
|
||||
const primary = this.#pool[0];
|
||||
const rest = this.#pool.slice(1);
|
||||
|
||||
await this.send(primary, {
|
||||
await this.#send(primary, {
|
||||
type: 'init',
|
||||
options: { appVersion, configDir, key },
|
||||
isPrimary: true,
|
||||
|
|
@ -182,7 +176,7 @@ export class MainSQL {
|
|||
|
||||
await Promise.all(
|
||||
rest.map(worker =>
|
||||
this.send(worker, {
|
||||
this.#send(worker, {
|
||||
type: 'init',
|
||||
options: { appVersion, configDir, key },
|
||||
isPrimary: false,
|
||||
|
|
@ -191,22 +185,22 @@ export class MainSQL {
|
|||
);
|
||||
})();
|
||||
|
||||
await this.onReady;
|
||||
await this.#onReady;
|
||||
|
||||
this.onReady = undefined;
|
||||
this.isReady = true;
|
||||
this.#onReady = undefined;
|
||||
this.#isReady = true;
|
||||
}
|
||||
|
||||
public pauseWriteAccess(): void {
|
||||
strictAssert(this.pauseWaiters == null, 'Already paused');
|
||||
strictAssert(this.#pauseWaiters == null, 'Already paused');
|
||||
|
||||
this.pauseWaiters = [];
|
||||
this.#pauseWaiters = [];
|
||||
}
|
||||
|
||||
public resumeWriteAccess(): void {
|
||||
const { pauseWaiters } = this;
|
||||
const pauseWaiters = this.#pauseWaiters;
|
||||
strictAssert(pauseWaiters != null, 'Not paused');
|
||||
this.pauseWaiters = undefined;
|
||||
this.#pauseWaiters = undefined;
|
||||
|
||||
for (const waiter of pauseWaiters) {
|
||||
waiter();
|
||||
|
|
@ -215,37 +209,39 @@ export class MainSQL {
|
|||
|
||||
public whenCorrupted(): Promise<Error> {
|
||||
const { promise, resolve } = explodePromise<Error>();
|
||||
this.errorResolvers.push({ kind: SqliteErrorKind.Corrupted, resolve });
|
||||
this.#errorResolvers.push({ kind: SqliteErrorKind.Corrupted, resolve });
|
||||
return promise;
|
||||
}
|
||||
|
||||
public whenReadonly(): Promise<Error> {
|
||||
const { promise, resolve } = explodePromise<Error>();
|
||||
this.errorResolvers.push({ kind: SqliteErrorKind.Readonly, resolve });
|
||||
this.#errorResolvers.push({ kind: SqliteErrorKind.Readonly, resolve });
|
||||
return promise;
|
||||
}
|
||||
|
||||
public async close(): Promise<void> {
|
||||
if (this.onReady) {
|
||||
if (this.#onReady) {
|
||||
try {
|
||||
await this.onReady;
|
||||
await this.#onReady;
|
||||
} catch (err) {
|
||||
this.logger?.error(`MainSQL close, failed: ${Errors.toLogFormat(err)}`);
|
||||
this.#logger?.error(
|
||||
`MainSQL close, failed: ${Errors.toLogFormat(err)}`
|
||||
);
|
||||
// Init failed
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (!this.isReady) {
|
||||
if (!this.#isReady) {
|
||||
throw new Error('Not initialized');
|
||||
}
|
||||
|
||||
await this.terminate({ type: 'close' });
|
||||
await this.onExit;
|
||||
await this.#terminate({ type: 'close' });
|
||||
await this.#onExit;
|
||||
}
|
||||
|
||||
public async removeDB(): Promise<void> {
|
||||
await this.terminate({ type: 'removeDB' });
|
||||
await this.#terminate({ type: 'removeDB' });
|
||||
}
|
||||
|
||||
public async sqlRead<Method extends keyof ServerReadableDirectInterface>(
|
||||
|
|
@ -261,16 +257,16 @@ export class MainSQL {
|
|||
// the same temporary table.
|
||||
const isPaging = PAGING_QUERIES.has(method);
|
||||
|
||||
const entry = isPaging ? this.pool.at(-1) : this.getWorker();
|
||||
const entry = isPaging ? this.#pool.at(-1) : this.#getWorker();
|
||||
strictAssert(entry != null, 'Must have a pool entry');
|
||||
|
||||
const { result, duration } = await this.send<SqlCallResult>(entry, {
|
||||
const { result, duration } = await this.#send<SqlCallResult>(entry, {
|
||||
type: 'sqlCall:read',
|
||||
method,
|
||||
args,
|
||||
});
|
||||
|
||||
this.traceDuration(method, duration);
|
||||
this.#traceDuration(method, duration);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
|
@ -285,63 +281,63 @@ export class MainSQL {
|
|||
duration: number;
|
||||
}>;
|
||||
|
||||
while (this.pauseWaiters != null) {
|
||||
while (this.#pauseWaiters != null) {
|
||||
const { promise, resolve } = explodePromise<void>();
|
||||
this.pauseWaiters.push(resolve);
|
||||
this.#pauseWaiters.push(resolve);
|
||||
// eslint-disable-next-line no-await-in-loop
|
||||
await promise;
|
||||
}
|
||||
|
||||
const primary = this.pool[0];
|
||||
const primary = this.#pool[0];
|
||||
|
||||
const { result, duration } = await this.send<SqlCallResult>(primary, {
|
||||
const { result, duration } = await this.#send<SqlCallResult>(primary, {
|
||||
type: 'sqlCall:write',
|
||||
method,
|
||||
args,
|
||||
});
|
||||
|
||||
this.traceDuration(method, duration);
|
||||
this.#traceDuration(method, duration);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
public startTrackingQueryStats(): void {
|
||||
if (this.shouldTrackQueryStats) {
|
||||
this.logQueryStats({});
|
||||
this.logger?.info('Resetting query stats');
|
||||
if (this.#shouldTrackQueryStats) {
|
||||
this.#logQueryStats({});
|
||||
this.#logger?.info('Resetting query stats');
|
||||
}
|
||||
this.resetQueryStats();
|
||||
this.shouldTrackQueryStats = true;
|
||||
this.#resetQueryStats();
|
||||
this.#shouldTrackQueryStats = true;
|
||||
}
|
||||
|
||||
public stopTrackingQueryStats(options: QueryStatsOptions): void {
|
||||
if (this.shouldTrackQueryStats) {
|
||||
this.logQueryStats(options);
|
||||
if (this.#shouldTrackQueryStats) {
|
||||
this.#logQueryStats(options);
|
||||
}
|
||||
this.queryStats = undefined;
|
||||
this.shouldTrackQueryStats = false;
|
||||
this.#queryStats = undefined;
|
||||
this.#shouldTrackQueryStats = false;
|
||||
}
|
||||
|
||||
private async send<Response>(
|
||||
async #send<Response>(
|
||||
entry: PoolEntry,
|
||||
request: WorkerRequest
|
||||
): Promise<Response> {
|
||||
if (request.type === 'sqlCall:read' || request.type === 'sqlCall:write') {
|
||||
if (this.onReady) {
|
||||
await this.onReady;
|
||||
if (this.#onReady) {
|
||||
await this.#onReady;
|
||||
}
|
||||
|
||||
if (!this.isReady) {
|
||||
if (!this.#isReady) {
|
||||
throw new Error('Not initialized');
|
||||
}
|
||||
}
|
||||
|
||||
const { seq } = this;
|
||||
const seq = this.#seq;
|
||||
// eslint-disable-next-line no-bitwise
|
||||
this.seq = (this.seq + 1) >>> 0;
|
||||
this.#seq = (this.#seq + 1) >>> 0;
|
||||
|
||||
const { promise: result, resolve, reject } = explodePromise<Response>();
|
||||
this.onResponse.set(seq, { resolve, reject });
|
||||
this.#onResponse.set(seq, { resolve, reject });
|
||||
|
||||
const wrappedRequest: WrappedWorkerRequest = {
|
||||
seq,
|
||||
|
|
@ -359,24 +355,24 @@ export class MainSQL {
|
|||
}
|
||||
}
|
||||
|
||||
private async terminate(request: WorkerRequest): Promise<void> {
|
||||
const primary = this.pool[0];
|
||||
const rest = this.pool.slice(1);
|
||||
async #terminate(request: WorkerRequest): Promise<void> {
|
||||
const primary = this.#pool[0];
|
||||
const rest = this.#pool.slice(1);
|
||||
|
||||
// Terminate non-primary workers first
|
||||
await Promise.all(rest.map(worker => this.send(worker, request)));
|
||||
await Promise.all(rest.map(worker => this.#send(worker, request)));
|
||||
|
||||
// Primary last
|
||||
await this.send(primary, request);
|
||||
await this.#send(primary, request);
|
||||
}
|
||||
|
||||
private onError(errorKind: SqliteErrorKind, error: Error): void {
|
||||
#onError(errorKind: SqliteErrorKind, error: Error): void {
|
||||
if (errorKind === SqliteErrorKind.Unknown) {
|
||||
return;
|
||||
}
|
||||
|
||||
const resolvers = new Array<(error: Error) => void>();
|
||||
this.errorResolvers = this.errorResolvers.filter(entry => {
|
||||
this.#errorResolvers = this.#errorResolvers.filter(entry => {
|
||||
if (entry.kind === errorKind) {
|
||||
resolvers.push(entry.resolve);
|
||||
return false;
|
||||
|
|
@ -389,76 +385,73 @@ export class MainSQL {
|
|||
}
|
||||
}
|
||||
|
||||
private resetQueryStats() {
|
||||
this.queryStats = { start: Date.now(), statsByQuery: new Map() };
|
||||
#resetQueryStats() {
|
||||
this.#queryStats = { start: Date.now(), statsByQuery: new Map() };
|
||||
}
|
||||
|
||||
private roundDuration(duration: number): number {
|
||||
#roundDuration(duration: number): number {
|
||||
return Math.round(100 * duration) / 100;
|
||||
}
|
||||
|
||||
private logQueryStats({
|
||||
maxQueriesToLog = 10,
|
||||
epochName,
|
||||
}: QueryStatsOptions) {
|
||||
if (!this.queryStats) {
|
||||
#logQueryStats({ maxQueriesToLog = 10, epochName }: QueryStatsOptions) {
|
||||
if (!this.#queryStats) {
|
||||
return;
|
||||
}
|
||||
const epochDuration = Date.now() - this.queryStats.start;
|
||||
const epochDuration = Date.now() - this.#queryStats.start;
|
||||
const sortedByCumulativeDuration = [
|
||||
...this.queryStats.statsByQuery.values(),
|
||||
...this.#queryStats.statsByQuery.values(),
|
||||
].sort((a, b) => (b.cumulative ?? 0) - (a.cumulative ?? 0));
|
||||
const cumulativeDuration = sortedByCumulativeDuration.reduce(
|
||||
(sum, stats) => sum + stats.cumulative,
|
||||
0
|
||||
);
|
||||
this.logger?.info(
|
||||
this.#logger?.info(
|
||||
`Top ${maxQueriesToLog} queries by cumulative duration (ms) over last ${epochDuration}ms` +
|
||||
`${epochName ? ` during '${epochName}'` : ''}: ` +
|
||||
`${sortedByCumulativeDuration
|
||||
.slice(0, maxQueriesToLog)
|
||||
.map(stats => {
|
||||
return (
|
||||
`${stats.queryName}: cumulative ${this.roundDuration(stats.cumulative)} | ` +
|
||||
`average: ${this.roundDuration(stats.cumulative / (stats.count || 1))} | ` +
|
||||
`max: ${this.roundDuration(stats.max)} | ` +
|
||||
`${stats.queryName}: cumulative ${this.#roundDuration(stats.cumulative)} | ` +
|
||||
`average: ${this.#roundDuration(stats.cumulative / (stats.count || 1))} | ` +
|
||||
`max: ${this.#roundDuration(stats.max)} | ` +
|
||||
`count: ${stats.count}`
|
||||
);
|
||||
})
|
||||
.join(' ||| ')}` +
|
||||
`; Total cumulative duration of all SQL queries during this epoch: ${this.roundDuration(cumulativeDuration)}ms`
|
||||
`; Total cumulative duration of all SQL queries during this epoch: ${this.#roundDuration(cumulativeDuration)}ms`
|
||||
);
|
||||
}
|
||||
|
||||
private traceDuration(method: string, duration: number): void {
|
||||
if (this.shouldTrackQueryStats) {
|
||||
if (!this.queryStats) {
|
||||
this.resetQueryStats();
|
||||
#traceDuration(method: string, duration: number): void {
|
||||
if (this.#shouldTrackQueryStats) {
|
||||
if (!this.#queryStats) {
|
||||
this.#resetQueryStats();
|
||||
}
|
||||
strictAssert(this.queryStats, 'has been initialized');
|
||||
let currentStats = this.queryStats.statsByQuery.get(method);
|
||||
strictAssert(this.#queryStats, 'has been initialized');
|
||||
let currentStats = this.#queryStats.statsByQuery.get(method);
|
||||
if (!currentStats) {
|
||||
currentStats = { count: 0, cumulative: 0, queryName: method, max: 0 };
|
||||
this.queryStats.statsByQuery.set(method, currentStats);
|
||||
this.#queryStats.statsByQuery.set(method, currentStats);
|
||||
}
|
||||
currentStats.count += 1;
|
||||
currentStats.cumulative += duration;
|
||||
currentStats.max = Math.max(currentStats.max, duration);
|
||||
}
|
||||
|
||||
if (this.shouldTimeQueries && !app.isPackaged) {
|
||||
const twoDecimals = this.roundDuration(duration);
|
||||
this.logger?.info(`MainSQL query: ${method}, duration=${twoDecimals}ms`);
|
||||
if (this.#shouldTimeQueries && !app.isPackaged) {
|
||||
const twoDecimals = this.#roundDuration(duration);
|
||||
this.#logger?.info(`MainSQL query: ${method}, duration=${twoDecimals}ms`);
|
||||
}
|
||||
if (duration > MIN_TRACE_DURATION) {
|
||||
strictAssert(this.logger !== undefined, 'Logger not initialized');
|
||||
this.logger.info(
|
||||
strictAssert(this.#logger !== undefined, 'Logger not initialized');
|
||||
this.#logger.info(
|
||||
`MainSQL: slow query ${method} duration=${Math.round(duration)}ms`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private createWorker(): CreateWorkerResultType {
|
||||
#createWorker(): CreateWorkerResultType {
|
||||
const scriptPath = join(app.getAppPath(), 'ts', 'sql', 'mainWorker.js');
|
||||
|
||||
const worker = new Worker(scriptPath);
|
||||
|
|
@ -466,15 +459,15 @@ export class MainSQL {
|
|||
worker.on('message', (wrappedResponse: WrappedWorkerResponse) => {
|
||||
if (wrappedResponse.type === 'log') {
|
||||
const { level, args } = wrappedResponse;
|
||||
strictAssert(this.logger !== undefined, 'Logger not initialized');
|
||||
this.logger[level](`MainSQL: ${format(...args)}`);
|
||||
strictAssert(this.#logger !== undefined, 'Logger not initialized');
|
||||
this.#logger[level](`MainSQL: ${format(...args)}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const { seq, error, errorKind, response } = wrappedResponse;
|
||||
|
||||
const pair = this.onResponse.get(seq);
|
||||
this.onResponse.delete(seq);
|
||||
const pair = this.#onResponse.get(seq);
|
||||
this.#onResponse.delete(seq);
|
||||
if (!pair) {
|
||||
throw new Error(`Unexpected worker response with seq: ${seq}`);
|
||||
}
|
||||
|
|
@ -483,7 +476,7 @@ export class MainSQL {
|
|||
const errorObj = new Error(error.message);
|
||||
errorObj.stack = error.stack;
|
||||
errorObj.name = error.name;
|
||||
this.onError(errorKind ?? SqliteErrorKind.Unknown, errorObj);
|
||||
this.#onError(errorKind ?? SqliteErrorKind.Unknown, errorObj);
|
||||
|
||||
pair.reject(errorObj);
|
||||
} else {
|
||||
|
|
@ -498,9 +491,9 @@ export class MainSQL {
|
|||
}
|
||||
|
||||
// Find first pool entry with minimal load
|
||||
private getWorker(): PoolEntry {
|
||||
let min = this.pool[0];
|
||||
for (const entry of this.pool) {
|
||||
#getWorker(): PoolEntry {
|
||||
let min = this.#pool[0];
|
||||
for (const entry of this.#pool) {
|
||||
if (min && min.load < entry.load) {
|
||||
continue;
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue