Apply a custom filter function `filter_func` to all tokens or ngrams (if `to_ngrams` is True).
        `filter_func` must accept a single parameter: a dictionary of structure `{<doc_label>: <tokens list>}`. It
        must return a dictionary with the same structure.

        This function can only be run on a single process, hence it could be slow for large corpora.

def apply_custom_filter(self, filter_func, to_ngrams=False):
        """
        Apply a custom filter function `filter_func` to all tokens or ngrams (if `to_ngrams` is True).
        `filter_func` must accept a single parameter: a dictionary of structure `{<doc_label>: <tokens list>}`. It
        must return a dictionary with the same structure.

        This function can only be run on a single process, hence it could be slow for large corpora.
        """

        # Because it is not possible to send a function to the workers, all tokens must be fetched from the workers
        # first and then the custom function is called and run in a single process (the main process). After that, the
        # filtered tokens are send back to the worker processes.

        if not callable(filter_func):
            raise ValueError('`filter_func` must be callable')

        self._require_tokens()

        if to_ngrams:
            self._require_ngrams()
            get_task = 'get_ngrams_with_worker_id'
            set_task = 'set_ngrams'
            set_task_param = 'ngrams'
            self._invalidate_workers_ngrams()
        else:
            get_task = 'get_tokens_with_worker_id'
            set_task = 'set_tokens'
            set_task_param = 'tokens'
            self._invalidate_workers_tokens()

        self._send_task_to_workers(get_task)

        docs_of_workers = {}
        for _ in range(self.n_workers):
            pair = self.results_queue.get()
            docs_of_workers[pair[0]] = pair[1]

        assert len(docs_of_workers) == self.n_workers

        tok = {}
        for docs in docs_of_workers.values():
            tok.update(docs)

        logger.info('applying custom filter function to tokens')
        new_tok = filter_func(tok)
        require_dictlike(new_tok)
        if set(new_tok.keys()) != set(tok.keys()):
            raise ValueError('the document labels and number of documents must stay unchanged during custom filtering')

        logger.debug('sending task `%s` to all workers' % set_task)
        for w_id, docs in docs_of_workers.items():
            new_w_docs = {dl: new_tok.pop(dl) for dl in docs}
            self.tasks_queues[w_id].put((set_task, {set_task_param: new_w_docs}))

        [q.join() for q in self.tasks_queues]

        return self