Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use less RAM in LdaMulticore #2300

Merged
merged 3 commits into from
Sep 26, 2020
+10 −6
Merged

Use less RAM in LdaMulticore #2300

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8edcf74
fix
horpto Jul 6, 2020
b3d16bc
Update CHANGELOG.md
mpenkov Sep 26, 2020
3e5108e
Merge branch 'develop' into ldamulticore-refactor-1
mpenkov Sep 26, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions gensim/models/ldamulticore.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -277,7 +277,7 @@ def process_result_queue(force=False):
self.log_perplexity(chunk, total_docs=lencorpus)

logger.info("training LDA model using %i processes", self.workers)
pool = Pool(self.workers, worker_e_step, (job_queue, result_queue,))
pool = Pool(self.workers, worker_e_step, (job_queue, result_queue, self))
for pass_ in range(self.passes):
queue_size, reallen = [0], 0
other = LdaState(self.eta, self.state.sstats.shape)
Expand All @@ -289,7 +289,7 @@ def process_result_queue(force=False):
# put the chunk into the workers' input job queue
while True:
try:
job_queue.put((chunk_no, chunk, self), block=False)
job_queue.put((chunk_no, chunk, self.state), block=False)
queue_size[0] += 1
logger.info(
"PROGRESS: pass %i, dispatched chunk #%i = documents up to #%i/%i, "
Expand All @@ -316,7 +316,7 @@ def process_result_queue(force=False):
pool.terminate()


def worker_e_step(input_queue, result_queue):
def worker_e_step(input_queue, result_queue, worker_lda):
"""Perform E-step for each job.
Parameters
Expand All @@ -326,17 +326,20 @@ def worker_e_step(input_queue, result_queue):
responsible for processing it.
result_queue : queue of :class:`~gensim.models.ldamodel.LdaState`
After the worker finished the job, the state of the resulting (trained) worker model is appended to this queue.
worker_lda : :class:`~gensim.models.ldamulticore.LdaMulticore`
LDA instance which performed e step
"""
logger.debug("worker process entering E-step loop")
while True:
logger.debug("getting a new job")
chunk_no, chunk, worker_lda = input_queue.get()
chunk_no, chunk, w_state = input_queue.get()
logger.debug("processing chunk #%i of %i documents", chunk_no, len(chunk))
worker_lda.state = w_state
worker_lda.sync_state()
worker_lda.state.reset()
worker_lda.do_estep(chunk) # TODO: auto-tune alpha?
del chunk
logger.debug("processed chunk, queuing the result")
result_queue.put(worker_lda.state)
del worker_lda # free up some memory
worker_lda.state = None
logger.debug("result put")