class_docstr: An optimized implementation of the LDA algorithm, able to harness the power of multicore CPUs.
  Follows the similar API as the parent class :class:`~gensim.models.ldamodel.LdaModel`.
class_docstr_tok: ["An","optimized","implementation","of","the","LDA","algorithm",",","able","to","harness","the","power","of","multicore","CPUs",".","Follows","the","similar","API","as","the","parent","class",":","class",":","`","~gensim.models.ldamodel.LdaModel","`","."]
docstr: Train the model with new documents, by EM-iterating over `corpus` until the topics converge
  (or until the maximum number of allowed iterations is reached).
  
  Train the model with new documents, by EM-iterating over the corpus until the topics converge, or until
  the maximum number of allowed iterations is reached. `corpus` must be an iterable. The E step is distributed
  into the several processes.
  
  Notes
  -----
  This update also supports updating an already trained model (`self`) with new documents from `corpus`;
  the two models are then merged in proportion to the number of old vs. new documents.
  This feature is still experimental for non-stationary input streams.
  
  For stationary input (no topic drift in new documents), on the other hand,
  this equals the online update of `'Online Learning for LDA' by Hoffman et al.`_
  and is guaranteed to converge for any `decay` in (0.5, 1].
  
  Parameters
  ----------
  corpus : {iterable of list of (int, float), scipy.sparse.csc}, optional
      Stream of document vectors or sparse matrix of shape (`num_documents`, `num_terms`) used to update the
      model.
  chunks_as_numpy : bool
      Whether each chunk passed to the inference step should be a np.ndarray or not. Numpy can in some settings
      turn the term IDs into floats, these will be converted back into integers in inference, which incurs a
      performance hit. For distributed computing it may be desirable to keep the chunks as `numpy.ndarray`.
docstr_tok: ["Train","the","model","with","new","documents",",","by","EM-iterating","over","`","corpus","`","until","the","topics","converge","(","or","until","the","maximum","number","of","allowed","iterations","is","reached",")",".","Train","the","model","with","new","documents",",","by","EM-iterating","over","the","corpus","until","the","topics","converge",",","or","until","the","maximum","number","of","allowed","iterations","is","reached",".","`","corpus","`","must","be","an","iterable",".","The","E","step","is","distributed","into","the","several","processes",".","Notes","--","--","-","This","update","also","supports","updating","an","already","trained","model","(","`","self","`",")","with","new","documents","from","`","corpus","`",";","the","two","models","are","then","merged","in","proportion","to","the","number","of","old","vs.","new","documents",".","This","feature","is","still","experimental","for","non-stationary","input","streams",".","For","stationary","input","(","no","topic","drift","in","new","documents",")",",","on","the","other","hand",",","this","equals","the","online","update","of","`","'Online","Learning","for","LDA","'","by","Hoffman","et","al.","`","_","and","is","guaranteed","to","converge","for","any","`","decay","`","in","(","0.5",",","1","]",".","Parameters","--","--","--","--","--","corpus",":","{","iterable","of","list","of","(","int",",","float",")",",","scipy.sparse.csc","}",",","optional","Stream","of","document","vectors","or","sparse","matrix","of","shape","(","`","num_documents","`",",","`","num_terms","`",")","used","to","update","the","model",".","chunks_as_numpy",":","bool","Whether","each","chunk","passed","to","the","inference","step","should","be","a","np.ndarray","or","not",".","Numpy","can","in","some","settings","turn","the","term","IDs","into","floats",",","these","will","be","converted","back","into","integers","in","inference",",","which","incurs","a","performance","hit",".","For","distributed","computing","it","may","be","desirable","to","keep","the","chunks","as","`","numpy.ndarray","`","."]
code: def update(self, corpus, chunks_as_numpy=False):
      """Train the model with new documents, by EM-iterating over `corpus` until the topics converge
      (or until the maximum number of allowed iterations is reached).
  
      Train the model with new documents, by EM-iterating over the corpus until the topics converge, or until
      the maximum number of allowed iterations is reached. `corpus` must be an iterable. The E step is distributed
      into the several processes.
  
      Notes
      -----
      This update also supports updating an already trained model (`self`) with new documents from `corpus`;
      the two models are then merged in proportion to the number of old vs. new documents.
      This feature is still experimental for non-stationary input streams.
  
      For stationary input (no topic drift in new documents), on the other hand,
      this equals the online update of `'Online Learning for LDA' by Hoffman et al.`_
      and is guaranteed to converge for any `decay` in (0.5, 1].
  
      Parameters
      ----------
      corpus : {iterable of list of (int, float), scipy.sparse.csc}, optional
          Stream of document vectors or sparse matrix of shape (`num_documents`, `num_terms`) used to update the
          model.
      chunks_as_numpy : bool
          Whether each chunk passed to the inference step should be a np.ndarray or not. Numpy can in some settings
          turn the term IDs into floats, these will be converted back into integers in inference, which incurs a
          performance hit. For distributed computing it may be desirable to keep the chunks as `numpy.ndarray`.
  
      """
      try:
          lencorpus = len(corpus)
      except TypeError:
          logger.warning(
              "input corpus stream has no len(); counting documents"
          )
          lencorpus = sum(1 for _ in corpus)
      if lencorpus == 0:
          logger.warning(
              "LdaMulticore.update() called with an empty corpus"
          )
          return
  
      self.state.numdocs += lencorpus
  
      if self.batch:
          updatetype = "batch"
          updateafter = lencorpus
      else:
          updatetype = "online"
          updateafter = self.chunksize * self.workers
      eval_every = self.eval_every or 0
      evalafter = min(lencorpus, eval_every * updateafter)
  
      updates_per_pass = max(1, lencorpus / updateafter)
      logger.info(
          "running %s LDA training, %s topics, %i passes over the supplied corpus of %i documents, "
          "updating every %i documents, evaluating every ~%i documents, "
          "iterating %ix with a convergence threshold of %f",
          updatetype,
          self.num_topics,
          self.passes,
          lencorpus,
          updateafter,
          evalafter,
          self.iterations,
          self.gamma_threshold,
      )
  
      if updates_per_pass * self.passes < 10:
          logger.warning(
              "too few updates, training might not converge; "
              "consider increasing the number of passes or iterations to improve accuracy"
          )
  
      job_queue = Queue(maxsize=2 * self.workers)
      result_queue = Queue()
  
      # rho is the "speed" of updating; TODO try other fncs
      # pass_ + num_updates handles increasing the starting t for each pass,
      # while allowing it to "reset" on the first pass of each update
      def rho():
          return pow(
              self.offset + pass_ + (self.num_updates / self.chunksize),
              -self.decay,
          )
  
      def process_result_queue(force=False):
          """
          Clear the result queue, merging all intermediate results, and update the
          LDA model if necessary.
  
          """
          merged_new = False
          while not result_queue.empty():
              other.merge(result_queue.get())
              queue_size[0] -= 1
              merged_new = True
  
          if (force and merged_new and queue_size[0] == 0) or (
              other.numdocs >= updateafter
          ):
              self.do_mstep(rho(), other, pass_ > 0)
              other.reset()
              if eval_every > 0 and (
                  force
                  or (self.num_updates / updateafter) % eval_every == 0
              ):
                  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, self)
      )
      for pass_ in range(self.passes):
          queue_size, reallen = [0], 0
          other = LdaState(self.eta, self.state.sstats.shape)
  
          chunk_stream = utils.grouper(
              corpus, self.chunksize, as_numpy=chunks_as_numpy
          )
          for chunk_no, chunk in enumerate(chunk_stream):
              reallen += len(
                  chunk
              )  # keep track of how many documents we've processed so far
  
              # put the chunk into the workers' input job queue
              while True:
                  try:
                      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, "
                          "outstanding queue size %i",
                          pass_,
                          chunk_no,
                          chunk_no * self.chunksize + len(chunk),
                          lencorpus,
                          queue_size[0],
                      )
                      break
                  except queue.Full:
                      # in case the input job queue is full, keep clearing the
                      # result queue, to make sure we don't deadlock
                      process_result_queue()
  
              process_result_queue()
          # endfor single corpus pass
  
          # wait for all outstanding jobs to finish
          while queue_size[0] > 0:
              process_result_queue(force=True)
  
          if reallen != lencorpus:
              raise RuntimeError(
                  "input corpus size changed during training (don't use generators as input)"
              )
      # endfor entire update
  
      pool.terminate()
  
file_path: gensim/gensim/models/ldamulticore.py
imports: import logging
  import queue
  from multiprocessing import Pool, Queue, cpu_count
  import numpy
  from gensim import utils
  from gensim.models.ldamodel import LdaModel, LdaState
label_desc: Use image node_id 2 for calling the LdaMulticore obj's underlying member method code with example usage: obj.update(corpus, chunks_as_numpy) and returns: pow