如何在多线程任务/时间图中可视化线程？

Question

我正在尝试使用matplotlib以有意义的方式绘制多线程代码的线程。我希望每个线程都可以用一种颜色显示。这样，该图将清楚显示哪些任务由哪个线程等执行。为了清楚起见，我想说（下图）黄色条是线程1执行的进程，红色条是线程2执行的进程，蓝色条是线程3执行的进程。

这似乎很难，我想出的最好的方法是在下面（请参见图和代码）。在示例中，我们有3个线程和12个任务（每个任务的持续时间在一定程度上是可变的）。线程1以黄色开始，线程2以红色开始，线程3以蓝色开始。我很乐意在整个图表中保持它们像这样的颜色编码。但是，我们看到的是线程1执行任务0、7和10，但是它将颜色从黄色切换为红色到红色。与线程2相同：它执行任务2、5、8和11，但是将颜色从红色切换为蓝色，再从蓝色切换为蓝色。与线程3相同。因此，颜色实际上以3个周期出现，并且与线程无关。我想让它们依赖于线程号，就像我之前说的那样，以使多线程图更有意义（因为现在不是）。

任何人都知道如何执行此操作吗？

import threading
import multiprocessing
import math
import numpy as np
import time
import matplotlib.pyplot as plt
import glob
from PIL import Image
import random
from random import sample
import string
from concurrent.futures import ThreadPoolExecutor


cpu_workers = 3
nSim = 12

def generate_bar_colors(cpu_workers):
    colors = ['red', 'gold', 'royalblue']
    return colors

def visualize_runtimes(results, title):
    colors = generate_bar_colors(cpu_workers)
    plt.rcParams["font.family"] = "Times New Roman"
    plt.rcParams['axes.axisbelow'] = True
    start,stop = np.array(results).T
    plt.barh(range(len(start)),stop-start,left=start, color=colors)
    plt.grid(axis='x', color= 'lightgrey')
    plt.title("Tasks", rotation='horizontal', fontsize=12, horizontalalignment="left", x=0)
    plt.xlabel("Seconds", fontsize=12, horizontalalignment='right', x=1.0)

def multithreading(func, args, workers):
    begin_time=time.time()
    with ThreadPoolExecutor(max_workers = workers) as executor:
        res = executor.map(func, args, [begin_time for i in range (len(args))])
    return list(res)

def simulation(i, base):
    start = time.time() - base
    print(str(threading.current_thread().getName()) + ': '+ str(i))
    time.sleep(math.cos(i)+i*0.1+1)
    stop = time.time() - base
    return start, stop


if __name__ == '__main__':
    visualize_runtimes(multithreading(simulation, i, cpu_workers), "Multi-threading")
    plt.savefig('foo.png', bbox_inches='tight')

plt.show()

Answer 1

实现此目的的一种方法（请参见下面的代码）。现在，每个线程都分配有颜色，并且很明显，计算是按需的多线程的（在这种情况下，3个线程正在执行12个任务）。

import threading

import math

from matplotlib.lines import Line2D


from concurrent.futures import ThreadPoolExecutor

from concurrent.futures import ProcessPoolExecutor

import matplotlib.pyplot as plt

from matplotlib.ticker import AutoMinorLocator, MultipleLocator, FuncFormatter
import time

import numpy as np




cpu_workers = 3

nSim = 12

i = range(nSim)



#Multi-threading function ----------------------------------------------------------------------------------------------

def multithreading(func, args, workers):

    with ThreadPoolExecutor(max_workers = workers) as executor:
        
        responses = executor.map(func, args)

    return list(responses)



#List of unique arguments in preserved order as in my_list -------------------------------------------------------------

def pres_uniq_list(my_list): #returns a unique list in preserved order

    seen = set()

    result = []

    for e in my_list:

        if e not in seen:

            result.append(e)

            seen.add(e)

    return result



#Get netto simulation start- and end-times as well as duration ---------------------------------------------------------

def sep_list_elements(list_of_lists, proc_start_t):

    start_values = [inner_list[0] for inner_list in list_of_lists]

    start_values = np.array(start_values) - proc_start_t


    end_values = [inner_list[1] for inner_list in list_of_lists]

    end_values = np.array(end_values) - proc_start_t

    return start_values, end_values



#Match colors with threads (one color per thread) ----------------------------------------------------------------------

def thread_colors(list_of_lists):

    thread_ids = [inner_list[2] for inner_list in list_of_lists]

    color_guide = ['red', 'royalblue', 'gold', 'darkgray', 'forestgreen', 'orangered', 'lightpink', 'teal']


    lookup = dict(zip(pres_uniq_list(thread_ids), color_guide))

    colors = [lookup[number] for number in thread_ids]

    return colors



#Graph legend to match with bars ---------------------------------------------------------------------------------------

def thread_legend(list_of_lists):

    thread_list = [inner_list[3] for inner_list in list_of_lists]

    color_guide = ['red', 'royalblue', 'gold', 'darkgray', 'forestgreen', 'orangered', 'lightpink', 'teal']


    lookup = dict(zip(pres_uniq_list(thread_list), color_guide))

    colors = [lookup[number] for number in thread_list]

    thread_legend = ([Line2D([0], [0], color=c, alpha=0.4, lw=4) for c in pres_uniq_list(colors)])

    names = [str('Thread-')+str(i) for i in range(cpu_workers)]

    return thread_legend, names



#Graph definition using MatPlotLib -------------------------------------------------------------------------------------

def graph_settings(start_t, end_t, title, colors, legend):


    plt.rcParams["font.family"] = "Times New Roman"

    plt.rcParams["font.size"] = 10

    plt.rcParams['axes.axisbelow'] = True

    plt.rcParams['axes.edgecolor'] = 'black'

    plt.rcParams['axes.linewidth'] = 0.8

    plt.rcParams['xtick.color'] = 'black'

    plt.rcParams['ytick.color'] = 'black'

    #plt.rcParams['font.weight']= 'heavy'



    fig = plt.figure(figsize=((12.0/2.54), (7.42/2.54)), facecolor='w', edgecolor='black')  #set (12,7.42) gulden snede but figsize is in inches

    fig, ax = plt.subplots()


    #ax.xaxis.set_major_locator(MultipleLocator(1.000))

    #ax.xaxis.set_minor_locator(AutoMinorLocator(4))

    #ax.yaxis.set_major_locator(MultipleLocator(1.000))

    #ax.yaxis.set_minor_locator(AutoMinorLocator(4))


    ax.set_xlim(-0.2, end_t[nSim-1])

    ax.set_ylim(-1, nSim)

    ax.spines['top'].set_visible(False) #to set color: ax.spines['top'].set_color('green')

    ax.spines['right'].set_visible(False)

    ax.spines['left'].set_smart_bounds(True)

    ax.spines['bottom'].set_smart_bounds(True)


    #ax.tick_params(which='major')

    #ax.tick_params(which='major', length=5)

    #ax.tick_params(which='minor', labelsize=10)

    #ax.tick_params(which='minor', length=2.5, labelsize=10, labelcolor='0.25')

    ax.margins(0.02) #margins in %, default is 5%

    ax.barh(range(len(start_t)), end_t-start_t, left=start_t, color=colors, alpha=0.4, edgecolor="none") #alpha adds vagueness to the bars

    ax.barh(range(len(start_t)), end_t-end_t+0.00001, left=end_t, color=colors, alpha=1, edgecolor="none") #alpha adds vagueness to the bars


    #w = pres_uniq_list(p.get_height() for p in ax.patches)
    #w_fl = float(i) for i in w]

    #ax.plot(end_t, range(nSim), linewidth=0, marker="o", markersize=w[0], color="grey")

    #print(end_t)

    #plt.grid(axis='x', color= 'lightgrey')

    #plt.ylabel("Tasks", fontsize=12, horizontalalignment='left', y=1.02, rotation='horizontal')

    ax.set_title(title, fontsize=14, fontweight="bold", horizontalalignment='center', y=1.04) #set title instead of y-label, bold does not work

    ax.annotate("Tasks", fontsize=12, xy=(0, 1), xytext=(0, 10), xycoords="axes fraction", textcoords="offset points", ha="right", ) #set ha="left" to have it above the axis

    ax.annotate(("Tasks: " + str(nSim) + " - Threads: " + str(cpu_workers)), fontsize=10, fontstyle='italic', xy=(0.5, 1), xytext=(0, 4), xycoords="axes fraction", textcoords="offset points", ha="center", color='gray' ) #set ha="left" to have it above the axis


    ax.set_xlabel("Time [s]", fontsize=12, horizontalalignment='right', x=1.0)
    #ax.set_suptitle(title, fontsize=16, fontweight='bold', y=1.005)

    thread_legend = reversed(legend[0])

    names = reversed(legend[1])



    leg = ax.legend(thread_legend, names, fancybox=True, loc='lower right', fontsize=10, edgecolor=None)

    #plt.legend(frameon=False)

    leg.get_frame().set_facecolor('none')

    leg.get_frame().set_edgecolor('none')


    plt.savefig('P1_5test.pdf', bbox_inches='tight')  #dpi=300 when png , bbox_inches='tight'

    plt.show()

    #plt.figure(figsize=(12, 10))

    #plt.savefig('foo1.pdf', , dpi = 300)

    return None



#Run code for if __name__ == '__main__' that links all functions -------------------------------------------------------

def plot_runtimes(workers, func, args, title):

    proc_start_t = time.time() # we track time from here

    runtimes = multithreading(func, args, workers) #collect task runtimes from function

    #net_runtimes = net_start_t(proc_start_t, runtimes) #extract net_runtimes from runtimes

    start_t, end_t = sep_list_elements(runtimes, proc_start_t) #seperate start_t and end_t

    colors = thread_colors(runtimes) #select thread colors

    legend = thread_legend(runtimes) #define thread legend

    graph_settings(start_t, end_t, title, colors, legend) #settings for the horizontal bar charts



#Tasks to perform ------------------------------------------------------------------------------------------------------

def simulation(i):

    #i = range(nSim)

    start_t = time.time()

    #print(str(i) + ': start')

    a = str(threading.current_thread().ident)

    b = str(threading.current_thread().getName())

    print(str(threading.current_thread().getName()) + ': '+ str(i))

    #print(str(i) + ': finish')

    time.sleep(math.cos(i)+i*0.1+1)

    end_t = time.time()

    return [start_t, end_t, a, b]

    #return a



#Definition ------------------------------------------------------------------------------------------------------------

if __name__ == '__main__':


    plot_runtimes(workers = cpu_workers,

                  func = simulation,

                  args = i,

                  title = "Multi-threading")