Skip to content

Commit

Permalink
Merge pull request #5304 from inception-project/refactoring/5303-Impr…
Browse files Browse the repository at this point in the history 
…ove-TaskMonitor-API

Issue #5303: Improve background task monitor
  • Loading branch information
@reckart
reckart authored Feb 20, 2025
2 parents 8351aad + 6a8fe34 commit ddbb7ec
Show file tree
Hide file tree
Showing 48 changed files with 1,335 additions and 985 deletions.
134 changes: 62 additions & 72 deletions ...java/de/tudarmstadt/ukp/clarin/webanno/agreement/task/CalculatePairwiseAgreementTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,6 @@
import de.tudarmstadt.ukp.clarin.webanno.api.casstorage.session.CasStorageSession;
import de.tudarmstadt.ukp.clarin.webanno.model.AnnotationDocument;
import de.tudarmstadt.ukp.clarin.webanno.model.AnnotationFeature;
import de.tudarmstadt.ukp.clarin.webanno.model.AnnotationLayer;
import de.tudarmstadt.ukp.clarin.webanno.model.SourceDocument;
import de.tudarmstadt.ukp.inception.documents.api.DocumentService;
import de.tudarmstadt.ukp.inception.scheduling.Task;
Expand Down Expand Up @@ -86,77 +85,76 @@ public void execute()
{
summary = new PairwiseAgreementResult(feature, traits);

var maxProgress = allAnnDocs.size();
var progress = 0;

var docs = allAnnDocs.keySet().stream() //
.sorted(comparing(SourceDocument::getName)) //
.toList();

for (var doc : docs) {
var monitor = getMonitor();
if (monitor.isCancelled()) {
break;
}

monitor.setProgressWithMessage(progress, maxProgress,
LogMessage.info(this, doc.getName()));

try (var session = CasStorageSession.openNested()) {
for (int m = 0; m < annotators.size(); m++) {
var annotator1 = annotators.get(m);
var maybeCas1 = LazyInitializer.<Optional<CAS>> builder()
.setInitializer(() -> loadCas(doc, annotator1, allAnnDocs)).get();

for (int n = 0; n < annotators.size(); n++) {
if (!(n < m)) {
// Triangle matrix mirrored
continue;
}

var annotator2 = annotators.get(n);

if ((CURATION_USER.equals(annotator1) || CURATION_USER.equals(annotator2))
&& !asList(CURATION_IN_PROGRESS, CURATION_FINISHED)
.contains(doc.getState())) {
LOG.trace("Skipping combination {}/{}@{}: {} not in a curation state",
annotator1, annotator2, doc, annotator1);
summary.mergeResult(annotator1, annotator2, AgreementSummary
.skipped(feature.getLayer().getName(), feature.getName()));
continue;
}

if (maybeCas1.get().isEmpty()) {
LOG.trace("Skipping combination {}/{}@{}: {} has no data", annotator1,
annotator2, doc, annotator1);
summary.mergeResult(annotator1, annotator2, AgreementSummary
.skipped(feature.getLayer().getName(), feature.getName()));
continue;
}

var maybeCas2 = LazyInitializer.<Optional<CAS>> builder()
.setInitializer(() -> loadCas(doc, annotator2, allAnnDocs)).get();
try (var progress = getMonitor().openScope("documents", allAnnDocs.size())) {
for (var doc : docs) {
if (getMonitor().isCancelled()) {
break;
}

if (maybeCas2.get().isEmpty()) {
LOG.trace("Skipping combination {}/{}@{}: {} has no data", annotator1,
annotator2, doc, annotator2);
summary.mergeResult(annotator1, annotator2, AgreementSummary
.skipped(feature.getLayer().getName(), feature.getName()));
continue;
progress.update(up -> up.increment() //
.addMessage(LogMessage.info(this, doc.getName())));

try (var session = CasStorageSession.openNested()) {
for (int m = 0; m < annotators.size(); m++) {
var annotator1 = annotators.get(m);
var maybeCas1 = LazyInitializer.<Optional<CAS>> builder()
.setInitializer(() -> loadCas(doc, annotator1, allAnnDocs)).get();

for (int n = 0; n < annotators.size(); n++) {
if (!(n < m)) {
// Triangle matrix mirrored
continue;
}

var annotator2 = annotators.get(n);

if ((CURATION_USER.equals(annotator1)
|| CURATION_USER.equals(annotator2))
&& !asList(CURATION_IN_PROGRESS, CURATION_FINISHED)
.contains(doc.getState())) {
LOG.trace(
"Skipping combination {}/{}@{}: {} not in a curation state",
annotator1, annotator2, doc, annotator1);
summary.mergeResult(annotator1, annotator2, AgreementSummary
.skipped(feature.getLayer().getName(), feature.getName()));
continue;
}

if (maybeCas1.get().isEmpty()) {
LOG.trace("Skipping combination {}/{}@{}: {} has no data",
annotator1, annotator2, doc, annotator1);
summary.mergeResult(annotator1, annotator2, AgreementSummary
.skipped(feature.getLayer().getName(), feature.getName()));
continue;
}

var maybeCas2 = LazyInitializer.<Optional<CAS>> builder()
.setInitializer(() -> loadCas(doc, annotator2, allAnnDocs))
.get();

if (maybeCas2.get().isEmpty()) {
LOG.trace("Skipping combination {}/{}@{}: {} has no data",
annotator1, annotator2, doc, annotator2);
summary.mergeResult(annotator1, annotator2, AgreementSummary
.skipped(feature.getLayer().getName(), feature.getName()));
continue;
}

var casMap = new LinkedHashMap<String, CAS>();
casMap.put(annotator1, maybeCas1.get().get());
casMap.put(annotator2, maybeCas2.get().get());
var res = AgreementSummary.of(measure.getAgreement(casMap));
summary.mergeResult(annotator1, annotator2, res);
}

var casMap = new LinkedHashMap<String, CAS>();
casMap.put(annotator1, maybeCas1.get().get());
casMap.put(annotator2, maybeCas2.get().get());
var res = AgreementSummary.of(measure.getAgreement(casMap));
summary.mergeResult(annotator1, annotator2, res);
}
}

progress++;
}
catch (Exception e) {
LOG.error("Unable to load data", e);
catch (Exception e) {
LOG.error("Unable to load data", e);
}
}
}
}
Expand Down Expand Up @@ -229,7 +227,6 @@ public static class Builder<T extends Builder<?>>
{
private List<String> annotators;
private DefaultAgreementTraits traits;
private AnnotationLayer layer;
private AnnotationFeature feature;
private AgreementMeasure<?> measure;
private Map<SourceDocument, List<AnnotationDocument>> allAnnDocs;
Expand All @@ -253,13 +250,6 @@ public T withTraits(DefaultAgreementTraits aTraits)
return (T) this;
}

@SuppressWarnings("unchecked")
public T withLayer(AnnotationLayer aLayer)
{
layer = aLayer;
return (T) this;
}

@SuppressWarnings("unchecked")
public T withFeature(AnnotationFeature aFeature)
{
Expand Down
64 changes: 26 additions & 38 deletions ...a/de/tudarmstadt/ukp/clarin/webanno/agreement/task/CalculatePerDocumentAgreementTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,6 @@
import de.tudarmstadt.ukp.clarin.webanno.model.AnnotationDocument;
import de.tudarmstadt.ukp.clarin.webanno.model.AnnotationDocumentState;
import de.tudarmstadt.ukp.clarin.webanno.model.AnnotationFeature;
import de.tudarmstadt.ukp.clarin.webanno.model.AnnotationLayer;
import de.tudarmstadt.ukp.clarin.webanno.model.SourceDocument;
import de.tudarmstadt.ukp.inception.documents.api.DocumentService;
import de.tudarmstadt.ukp.inception.scheduling.Task;
Expand Down Expand Up @@ -88,45 +87,42 @@ public void execute()
{
summary = new PerDocumentAgreementResult(feature, traits);

var maxProgress = allAnnDocs.size();
var progress = 0;

var docs = allAnnDocs.keySet().stream() //
.sorted(comparing(SourceDocument::getName)) //
.toList();

for (var doc : docs) {
var monitor = getMonitor();
if (monitor.isCancelled()) {
break;
}
try (var progress = getMonitor().openScope("documents", allAnnDocs.size())) {
for (var doc : docs) {
if (getMonitor().isCancelled()) {
break;
}

progress.update(up -> up.increment() //
.addMessage(LogMessage.info(this, doc.getName())));

monitor.setProgressWithMessage(progress, maxProgress,
LogMessage.info(this, doc.getName()));
try (var session = CasStorageSession.openNested()) {
var casMap = new LinkedHashMap<String, CAS>();
for (var annDoc : allAnnDocs.get(doc)) {
var dataOwner = annDoc.getUser();
if (!annotators.contains(dataOwner)) {
continue;
}

try (var session = CasStorageSession.openNested()) {
var casMap = new LinkedHashMap<String, CAS>();
for (var annDoc : allAnnDocs.get(doc)) {
var dataOwner = annDoc.getUser();
if (!annotators.contains(dataOwner)) {
continue;
casMap.put(dataOwner, loadCas(annDoc.getDocument(), dataOwner));
}

casMap.put(dataOwner, loadCas(annDoc.getDocument(), dataOwner));
}
if (annotators.contains(CURATION_USER)) {
casMap.put(CURATION_USER, loadCas(doc, CURATION_USER));
}

if (annotators.contains(CURATION_USER)) {
casMap.put(CURATION_USER, loadCas(doc, CURATION_USER));
LOG.trace("Calculating agreement on {} for [{}] annotators", doc,
casMap.size());
var agreementResult = AgreementSummary.of(measure.getAgreement(casMap));
summary.mergeResult(doc, agreementResult);
}
catch (Exception e) {
LOG.error("Unable to load data", e);
}

LOG.trace("Calculating agreement on {} for [{}] annotators", doc, casMap.size());
var agreementResult = AgreementSummary.of(measure.getAgreement(casMap));
summary.mergeResult(doc, agreementResult);

progress++;
}
catch (Exception e) {
LOG.error("Unable to load data", e);
}
}
}
Expand Down Expand Up @@ -184,7 +180,6 @@ public static class Builder<T extends Builder<?>>
{
private List<String> annotators;
private DefaultAgreementTraits traits;
private AnnotationLayer layer;
private AnnotationFeature feature;
private AgreementMeasure<?> measure;
private Map<SourceDocument, List<AnnotationDocument>> allAnnDocs;
Expand All @@ -209,13 +204,6 @@ public T withTraits(DefaultAgreementTraits aTraits)
return (T) this;
}

@SuppressWarnings("unchecked")
public T withLayer(AnnotationLayer aLayer)
{
layer = aLayer;
return (T) this;
}

@SuppressWarnings("unchecked")
public T withFeature(AnnotationFeature aFeature)
{
Expand Down
91 changes: 41 additions & 50 deletions ...c/main/java/de/tudarmstadt/ukp/inception/assistant/documents/UpdateDocumentIndexTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,9 @@
import static de.tudarmstadt.ukp.inception.assistant.documents.DocumentQueryService.FIELD_SOURCE_DOC_COMPLETE;
import static de.tudarmstadt.ukp.inception.assistant.documents.DocumentQueryService.FIELD_SOURCE_DOC_ID;
import static de.tudarmstadt.ukp.inception.assistant.documents.DocumentQueryService.FIELD_TEXT;
import static de.tudarmstadt.ukp.inception.scheduling.ProgressScope.SCOPE_DOCUMENTS;
import static de.tudarmstadt.ukp.inception.scheduling.ProgressScope.SCOPE_UNITS;
import static de.tudarmstadt.ukp.inception.scheduling.TaskState.CANCELLED;
import static de.tudarmstadt.ukp.inception.scheduling.TaskState.COMPLETED;
import static de.tudarmstadt.ukp.inception.scheduling.TaskState.RUNNING;
import static java.lang.Math.floorDiv;
import static java.time.Duration.ofSeconds;
import static org.apache.commons.collections4.ListUtils.partition;
Expand Down Expand Up @@ -138,41 +138,38 @@ public void execute() throws Exception
}
}

var toProcess = documentsToUnindex.size() + documentsToIndex.size() * 100;
var processed = 0;
monitor.setStateAndProgress(RUNNING, processed * 100, toProcess);

for (var sourceDocumentId : documentsToUnindex) {
if (monitor.isCancelled()) {
monitor.setState(CANCELLED);
break;
}

monitor.setProgressWithMessage(processed * 100, toProcess,
LogMessage.info(this, "Unindexing..."));
unindexDocument(index, sourceDocumentId);
processed++;
}

try (var session = CasStorageSession.openNested()) {
for (var sourceDocument : documentsToIndex.values()) {
try (var progress = getMonitor().openScope(SCOPE_DOCUMENTS,
documentsToUnindex.size() + documentsToIndex.size())) {
progress.update(up -> up.addMessage(LogMessage.info(this, "Unindexing...")));
for (var sourceDocumentId : documentsToUnindex) {
if (monitor.isCancelled()) {
monitor.setState(CANCELLED);
monitor.update(up -> up.setState(CANCELLED));
break;
}

monitor.setProgressWithMessage(processed * 100, toProcess,
LogMessage.info(this, "Indexing: %s", sourceDocument.getName()));
indexDocument(index, chunker, sourceDocument);
processed++;
progress.update(up -> up.increment());
unindexDocument(index, sourceDocumentId);
}

try (var session = CasStorageSession.openNested()) {
for (var sourceDocument : documentsToIndex.values()) {
if (monitor.isCancelled()) {
monitor.update(up -> up.setState(CANCELLED));
break;
}

progress.update(up -> up.increment() //
.addMessage(LogMessage.info(this, "Indexing: %s",
sourceDocument.getName())));
indexDocument(index, chunker, sourceDocument);
}
}
}

if (!monitor.isCancelled()) {
monitor.setStateAndProgress(COMPLETED, processed * 100, toProcess);
index.getIndexWriter().commit();
// We can probably live with a partial index, so we do not roll back if
// cancelled
if (!monitor.isCancelled()) {
index.getIndexWriter().commit();
// We can probably live with a partial index, so we do not roll back if
// cancelled
}
}
}
}
Expand Down Expand Up @@ -204,27 +201,21 @@ private void indexDocument(LuceneIndexPool.PooledIndex aIndex, Chunker<CAS> aChu
var chunks = aChunker.process(cas);
var batches = partition(chunks, properties.getEmbedding().getBatchSize());

var totalBatches = batches.size();
var processedBatches = 0;
var chunksSeen = 0;
var progressOffset = getMonitor().getProgress();
for (var batch : batches) {
try {
var docEmbeddings = embeddingService.embed(Chunk::text, batch);
for (var embedding : docEmbeddings) {
indexChunks(aIndex, aSourceDocument, embedding);
try (var progress = getMonitor().openScope(SCOPE_UNITS, chunks.size())) {
for (var batch : batches) {
progress.update(up -> up.increment(batch.size()));

try {
var docEmbeddings = embeddingService.embed(Chunk::text, batch);
for (var embedding : docEmbeddings) {
indexChunks(aIndex, aSourceDocument, embedding);
}
}
catch (IOException e) {
LOG.error("Error indexing document {} chunks {}-{} ", aSourceDocument,
progress.getProgress(), progress.getProgress() + batch.size(), e);
}
}
catch (IOException e) {
LOG.error("Error indexing document {} chunks {}-{} ", aSourceDocument,
chunksSeen, chunksSeen + batch.size(), e);
}

getMonitor().setStateAndProgress(RUNNING,
progressOffset + floorDiv(processedBatches * 100, totalBatches));

chunksSeen += batch.size();
processedBatches++;
}

markDocumentAsIndexed(aIndex, aSourceDocument);
Expand Down
Loading

0 comments on commit ddbb7ec

Please sign in to comment.