-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathsync.ts
More file actions
233 lines (197 loc) · 5.72 KB
/
sync.ts
File metadata and controls
233 lines (197 loc) · 5.72 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
export interface Resource<T> {
acquire(): Promise<T>
release(): void
}
export function promise<T>(): [
Promise<T>,
(value?: T | PromiseLike<T>) => void,
(reason?: any) => void,
] {
let resolve: ((value?: T | PromiseLike<T>) => void) | undefined
let reject: ((reason?: any) => void) | undefined
const prom = new Promise<T>((res, rej) => {
resolve = res
reject = rej
})
if (resolve === undefined || reject === undefined)
throw new Error('broken promise implementation')
return [prom, resolve, reject]
}
export function lock<T, R>(resource: Resource<T>, fn: (value: T) => Promise<R>): Promise<R> {
return resource.acquire().then(fn).finally(() => resource.release())
}
export function all<T>(ps: Promise<T>[]): Promise<T[]> {
return new Promise((res, rej) => {
const r: T[] = Array(ps.length)
let done = 0
for (let i = 0; i < ps.length; ++i) {
ps[i].then(v => {
r[i] = v
++done
if (done === ps.length)
res(r)
}, rej)
}
})
}
export function race<T>(ps: Promise<T>[]): Promise<T> {
return new Promise((res, rej) => {
for (const p of ps) {
p.then(res, rej)
}
})
}
export class EofError extends Error {
constructor() {
super('EOF reached')
}
}
export class Sync<T> implements PromiseLike<T> {
private _value: Promise<T>
private _resolve: (value?: T | PromiseLike<T>) => void
private _reject: (reason?: any) => void
private _status: 'pending' | 'resolved' | 'rejected' = 'pending'
constructor() {
[this._value, this._resolve, this._reject] = promise()
}
async then<R1 = T, R2 = never>(
onfulfilled?: ((value: T) => R1 | PromiseLike<R1>) | null | undefined,
onrejected?: ((reason: any) => R2 | PromiseLike<R2>) | null | undefined,
): Promise<R1 | R2> {
return this._value.then(onfulfilled, onrejected)
}
resolve(value?: T | PromiseLike<T>): void {
if (this._status !== 'pending')
return
this._resolve(value)
this._status = 'resolved'
}
reject(reason: any): void {
if (this._status !== 'pending')
return
this._reject(reason)
this._status = 'rejected'
}
get status(): 'pending' | 'resolved' | 'rejected' {
return this._status
}
}
export class Wait<T> {
private wait: Sync<T>[] = []
async block(): Promise<T> {
const sync = new Sync<T>()
this.wait.push(sync)
return sync
}
// returns true if a promise is resolved
unblock(value: T): boolean {
const sync = this.wait.shift()
if (sync === undefined)
return false
sync.resolve(value)
return true
}
broadcast(value: T): void {
for (const sync of this.wait)
sync.resolve(value)
this.wait = []
}
close(reason: any): void {
for (const sync of this.wait)
sync.reject(reason)
this.wait = []
}
}
export class Queue<T> implements AsyncIterable<T>, AsyncIterator<T, void> {
private buffer: T[] = []
private wait = new Wait<T>()
private _closed = false
private _end = new Sync<void>()
push(...values: T[]): void {
if (this._closed)
throw new Error('queue closed')
for (const value of values) {
if (!this.wait.unblock(value))
this.buffer.push(value)
}
}
close(): void {
if (this._closed)
throw new Error('already closed')
this._closed = true
this.wait.close(new EofError())
this._end.resolve()
}
async pop(): Promise<T> {
if (this._closed)
throw new EofError()
const item = this.buffer.shift()
if (item !== undefined)
return item
return this.wait.block()
}
async end(): Promise<void> {
return this._end
}
async next(): Promise<IteratorResult<T, void>> {
try {
return { done: false, value: await this.pop() }
} catch (e) {
if (e instanceof EofError)
return { done: true, value: undefined }
throw e
}
}
[Symbol.asyncIterator](): AsyncIterator<T> {
return this
}
get closed(): boolean { return this._closed }
get length(): number { return this.buffer.length }
}
export class Lock implements Resource<void> {
private wait = new Wait<void>()
private used = false
async acquire(): Promise<void> {
if (this.used)
await this.wait.block()
else
this.used = true
}
release(): void {
if (!this.used)
throw new Error('releasing an unused lock')
if (!this.wait.unblock())
this.used = false
}
}
export class Sem implements Resource<void> {
private wait = new Wait<void>()
constructor(private count: number) { }
async acquire(): Promise<void> {
if (this.count === 0)
await this.wait.block()
else
--this.count
}
release(): void {
if (!this.wait.unblock())
++this.count
}
P = this.acquire
V = this.release
}
export function streamLines(stream: NodeJS.ReadableStream): Queue<string> {
const queue = new Queue<string>()
let buffer = ''
stream.setEncoding('utf8')
stream.on('data', (chunk: string) => {
const lines = chunk.split('\n')
if (lines.length === 0)
throw 'broken String.prototype.split'
lines[0] = buffer + lines[0]
buffer = lines.pop()!
queue.push(...lines)
})
stream.on('end', () => queue.close())
return queue
}