<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def run_pipelines(pipeline_id_pattern, root_dir, use_cache=True, dirty=False, force=False, concurrency=1, verbose_logs=True, progress_cb=None, slave=False):
 """Run a pipeline by pipeline-id. pipeline-id supports the '%' wildcard for any-suffix matching. Use 'all' or '%' for running all pipelines"""

    with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency,
                                               thread_name_prefix='T') as executor:
        try:
            results = []
            pending_futures = set()
            done_futures = set()
            finished_futures = []
            progress_thread = None
            progress_queue = None
            status_manager = status_mgr(root_dir)

            if progress_cb is not None:
                progress_queue = Queue()
                progress_thread = threading.Thread(target=progress_report_handler, args=(progress_cb, progress_queue))
                progress_thread.start()

            all_specs = specs_to_execute(pipeline_id_pattern, root_dir, status_manager, force, dirty, results)

            while True:

                done = None
                if len(done_futures) > 0:
                    done = done_futures.pop()
                    finished_futures.append(done)
                    done = done.result()[0]

                try:
                    spec = all_specs.send(done)
                except StopIteration:
                    spec = None

                if spec is None:
                    # Wait for all runners to idle...
                    if len(done_futures) == 0:
                        if len(pending_futures) > 0:
                            done_futures, pending_futures = \
                                concurrent.futures.wait(pending_futures,
                                                        return_when=concurrent.futures.FIRST_COMPLETED)
                            continue
                        else:
                            break
                    else:
                        continue

                if len(spec.validation_errors) > 0:
                    results.append(
                        ExecutionResult(spec.pipeline_id,
                                        False,
                                        {},
                                        ['init'] + list(map(str, spec.validation_errors)))
                    )
                    continue

                if slave:
                    ps = status_manager.get(spec.pipeline_id)
                    ps.init(spec.pipeline_details,
                            spec.source_details,
                            spec.validation_errors,
                            spec.cache_hash)
                    eid = gen_execution_id()
                    if ps.queue_execution(eid, 'manual'):
                        success, stats, errors = \
                            execute_pipeline(spec, eid,
                                             use_cache=use_cache)
                        results.append(ExecutionResult(
                            spec.pipeline_id,
                            success,
                            stats,
                            errors
                        ))
                    else:
                        results.append(
                            ExecutionResult(spec.pipeline_id,
                                            False,
                                            None,
                                            ['Already Running'])
                        )

                else:
                    f = executor.submit(remote_execute_pipeline,
                                        spec,
                                        root_dir,
                                        use_cache,
                                        verbose_logs,
                                        progress_queue)
                    pending_futures.add(f)

            for f in finished_futures:
                ret = f.result()
                results.append(ExecutionResult(*ret))

        except KeyboardInterrupt:
            pass
        finally:
            if slave:
                finalize()

            if progress_thread is not None:
                progress_queue.put(None)
                progress_thread.join()

    return results