private void deploySelectorVertex(final JobDescriptor job, final LocalSelectorVertex pv) throws IOException {
        final PullModuleThreadUnit unit = unitController.createSelectorUnit("SelectorThread", pv.getCpuAffinityIds());
        final SelectorContext<?, ?> context = (SelectorContext<?, ?>) unit.getThreadContext();

        for (int i = 0; i < pv.getSelectorInputCount(); i++) {
            int inputThreadLocalId = unitController.createThreadLocalInputQueue(unit);
            unitController.addNewInputQueue(context, "SelectorInput" + i, i, inputThreadLocalId);
            unitController.addThreadInputRoute(unit, inputThreadLocalId, context, i);
            pv.setInputQueueThreadLocalId(i, inputThreadLocalId);
        }
        for (int i = 0; i < pv.getSelectorOutputCout(); i++) {
            int newThreadLocalId = unitController.createThreadLocalOutputQueue(unit);
            unitController.addNewOutputQueue(context, "SelectorOutput" + i, i, newThreadLocalId);
            pv.setOutputQueueThreadLocalId(i, newThreadLocalId);

            int port = context.startAccepting(i, pv.getBindAddress());
            pv.setPort(i, port);
        }

        threadUnits.get(job).addNewThreadUnit(unit);
        pv.setContext(context);
        pv.setThreadUnit(unit);

        // // We have to start the unit to accept socket connections from remote workers.
        // unitController.startThreadUnit(unit);
    }