Adding questions using worker pools

This commit is contained in:
mrfry 2020-12-19 10:32:15 +01:00
parent d8695682f7
commit 4681ea3791
5 changed files with 50 additions and 17 deletions

View file

@ -45,7 +45,7 @@ fi
echo "Result:" echo "Result:"
../bin/hr.sh ../bin/hr.sh
res=$(curl -L -s -X GET "${url}/ask?q=${q}&data=${data}&subj=${subj}") res=$(curl -H "Content-Type: application/json" -L -s -X GET "${url}/ask?q=${q}&data=${data}&subj=${subj}")
echo "$res" | jq echo "$res" | jq
if [ "$?" -ne 0 ]; then if [ "$?" -ne 0 ]; then
echo "jq error" echo "jq error"

View file

@ -1067,16 +1067,31 @@ function GetApp(): ModuleType {
if (req.query.q && req.query.data) { if (req.query.q && req.query.data) {
const subj: any = req.query.subj || '' const subj: any = req.query.subj || ''
const question = req.query.q const question = req.query.q
const recData: any = req.query.data let recData: any = req.query.data
if (typeof recData === 'string') {
try {
recData = JSON.parse(recData)
} catch (err) {
logger.Log(
'Error parsing recData in /ask!',
logger.GetColor('redbg')
)
console.error(err)
}
}
const promises = [] const promises = []
questionDbs.map((qdb, i) => { questionDbs.map((qdb) => {
promises.push( promises.push(
doALongTask(i, { doALongTask({
type: 'work', type: 'work',
index: i, data: {
data: { qdb: qdb.data, question, subjName: subj, recData }, qdb: qdb.data,
question,
subjName: subj,
questionData: recData,
},
}) })
) )
}) })
@ -1102,11 +1117,11 @@ function GetApp(): ModuleType {
success: true, success: true,
}) })
logger.DebugLog( logger.DebugLog(
`Question result length: ${result.length}`, `Question result length: ${mergedResult.length}`,
'ask', 'ask',
1 1
) )
logger.DebugLog(result, 'ask', 2) logger.DebugLog(mergedResult, 'ask', 2)
} catch (err) { } catch (err) {
console.error(err) console.error(err)
logger.Log( logger.Log(

View file

@ -21,7 +21,8 @@ const recDataFile = './stats/recdata'
const dataLockFile = './data/lockData' const dataLockFile = './data/lockData'
import logger from '../utils/logger' import logger from '../utils/logger'
import { searchData, createQuestion } from '../utils/classes' import { createQuestion } from '../utils/classes'
import { doALongTask } from './workerPool'
import idStats from '../utils/ids' import idStats from '../utils/ids'
import utils from '../utils/utils' import utils from '../utils/utils'
import { SearchResult, addQuestion, getSubjNameWithoutYear } from './classes' import { SearchResult, addQuestion, getSubjNameWithoutYear } from './classes'
@ -166,7 +167,14 @@ function processIncomingRequestUsingDb(
logger.DebugLog(currentQuestion, 'actions', 3) logger.DebugLog(currentQuestion, 'actions', 3)
recievedQuestions.push(currentQuestion) recievedQuestions.push(currentQuestion)
questionSearchPromises.push( questionSearchPromises.push(
searchData(qdb, currentQuestion, recievedData.subj) doALongTask({
type: 'work',
data: {
qdb: qdb.data,
question: currentQuestion,
subjName: recievedData.subj,
},
})
) )
}) })

View file

@ -585,7 +585,6 @@ const workerTs = (file: string, wkOpts: any) => {
if (!isMainThread) { if (!isMainThread) {
const workerIndex = workerData.workerIndex const workerIndex = workerData.workerIndex
// TODO: check if thread independent
logger.Log( logger.Log(
`[THREAD #${workerIndex}]: Worker ${workerIndex} reporting for duty` `[THREAD #${workerIndex}]: Worker ${workerIndex} reporting for duty`
@ -595,7 +594,11 @@ if (!isMainThread) {
if (msg.type === 'work') { if (msg.type === 'work') {
const { qdb, subjName, question, questionData } = msg.data const { qdb, subjName, question, questionData } = msg.data
const index = msg.index const index = msg.index
console.log(`[THREAD #${workerIndex}]: staring work on ${index}`) console.log(
`[THREAD #${workerIndex}]: staring work${
!isNaN(index) ? ` on job index #${index}` : ''
}`
)
let searchResult = null let searchResult = null
try { try {
@ -607,12 +610,18 @@ if (!isMainThread) {
// ONDONE: // ONDONE:
parentPort.postMessage({ parentPort.postMessage({
msg: `From thread #${workerIndex}: job ${index} done`, msg: `From thread #${workerIndex}: job ${
!isNaN(index) ? `#${index}` : ''
}done`,
workerIndex: workerIndex, workerIndex: workerIndex,
result: searchResult, result: searchResult,
}) })
console.log(`[THREAD #${workerIndex}]: Work ${index} done!`) console.log(
`[THREAD #${workerIndex}]: Work ${
!isNaN(index) ? `#${index}` : ''
}done!`
)
} else if (msg.type === 'update') { } else if (msg.type === 'update') {
if (msg.data.workerIndex !== workerIndex) { if (msg.data.workerIndex !== workerIndex) {
// TODO // TODO

View file

@ -18,7 +18,7 @@ let workers: any = null
// console.log('[MSGFROMCLIENT]', res) // console.log('[MSGFROMCLIENT]', res)
// }) // })
export function doALongTask(i: Number, obj: any): Promise<any> { export function doALongTask(obj: any): Promise<any> {
return new Promise((resolve) => { return new Promise((resolve) => {
pool pool
.acquire() .acquire()
@ -47,11 +47,12 @@ export function initWorkerPool(): void {
workers = [] workers = []
const factory = { const factory = {
create: function() { create: function() {
const worker = getAWorker(workers.length) const currInd = workers.length
const worker = getAWorker(currInd)
workers.push(worker) workers.push(worker)
return { return {
worker: worker, worker: worker,
index: workers.length, index: currInd,
} }
}, },
destroy: function(client) { destroy: function(client) {