()
| 2375 | } |
| 2376 | |
| 2377 | func (t *BleveDestPartition) submitAsyncBatchRequestLOCKED() (bool, error) { |
| 2378 | // fetch the needed parameters and remain unlocked until requestCh |
| 2379 | // is ready to accommodate this request |
| 2380 | bindex := t.bindex |
| 2381 | seqMaxBuf := make([]byte, 8) |
| 2382 | binary.BigEndian.PutUint64(seqMaxBuf, t.seqMax) |
| 2383 | t.batch.SetInternal(t.partitionBytes, seqMaxBuf) |
| 2384 | batch := t.batch |
| 2385 | t.batch = t.bindex.NewBatch() |
| 2386 | p := t.partition |
| 2387 | batchReqChs := t.bdest.batchReqChs |
| 2388 | stopCh := t.bdest.stopCh |
| 2389 | t.m.Unlock() |
| 2390 | |
| 2391 | // ensure that batch requests from a given partition always goes |
| 2392 | // to the same worker queue so that the order of seq numbers are maintained |
| 2393 | partition, err := strconv.Atoi(p) |
| 2394 | if err != nil { |
| 2395 | log.Errorf("pindex_bleve: submitAsyncBatchRequestLOCKED over `%v`, err: %v", |
| 2396 | bindex.Name(), err) |
| 2397 | t.m.Lock() |
| 2398 | return false, err |
| 2399 | } |
| 2400 | |
| 2401 | reqChIndex := partition % asyncBatchWorkerCount |
| 2402 | br := &batchRequest{bdp: t, bindex: bindex, |
| 2403 | batch: batch, |
| 2404 | } |
| 2405 | select { |
| 2406 | case <-stopCh: |
| 2407 | log.Printf("pindex_bleve: submitAsyncBatchRequestLOCKED stopped for `%v`", |
| 2408 | bindex.Name()) |
| 2409 | t.m.Lock() |
| 2410 | return false, t.lastAsyncBatchErr |
| 2411 | |
| 2412 | case batchReqChs[reqChIndex] <- br: |
| 2413 | } |
| 2414 | |
| 2415 | // acquire lock |
| 2416 | t.m.Lock() |
| 2417 | return false, t.lastAsyncBatchErr |
| 2418 | } |
| 2419 | |
| 2420 | func (t *BleveDestPartition) setLastAsyncBatchErr(err error) { |
| 2421 | t.m.Lock() |
no test coverage detected