import os import sys import time import shutil import dataclasses from multiprocessing import Process,Queue,Value,cpu_count CC="riscv64-unknown-elf-gcc" OBJCPY="riscv64-unknown-elf-objcopy" OBJDUMP="riscv64-unknown-elf-objdump" CFLAG=[ "-march=rv32i_zicsr", "-mabi=ilp32", "-ffunction-sections", "-fdata-sections", "-ffast-math", "-fno-common", "-fno-builtin-printf", "-Wall", "-Werror", "-g", "-O0", "-fno-omit-frame-pointer", "-msave-restore" ] DEF=[] INC=[ "-I./riscv/rtthread/libcpu/risc-v/common", "-I./riscv/rtthread/include", "-I./riscv/startup", "-I./riscv/rtthread/components/drivers/include" ] SRC=[ "riscv/startup/main.c", "riscv/startup/test.c", # "riscv/startup/interrupt.c", "riscv/startup/start.S", # "riscv/startup/trap.S", "riscv/startup/rtthread_irq.c", "riscv/rtthread/libcpu/risc-v/common/atomic_riscv.c", "riscv/rtthread/libcpu/risc-v/common/context_gcc.S", "riscv/rtthread/libcpu/risc-v/common/cpuport.c", "riscv/rtthread/libcpu/risc-v/common/interrupt_gcc.S", "riscv/rtthread/libcpu/risc-v/common/trap_common.c", "riscv/rtthread/src/clock.c", "riscv/rtthread/src/components.c", "riscv/rtthread/src/cpu.c", "riscv/rtthread/src/idle.c", "riscv/rtthread/src/ipc.c", "riscv/rtthread/src/irq.c", "riscv/rtthread/src/kservice.c", "riscv/rtthread/src/mem.c", "riscv/rtthread/src/memheap.c", "riscv/rtthread/src/mempool.c", "riscv/rtthread/src/object.c", "riscv/rtthread/src/scheduler_comm.c", # "riscv/rtthread/src/scheduler_mp.c", "riscv/rtthread/src/scheduler_up.c", "riscv/rtthread/src/signal.c", "riscv/rtthread/src/slab.c", "riscv/rtthread/src/thread.c", "riscv/rtthread/src/timer.c", "riscv/rtthread/src/klibc/kstdio.c", "riscv/rtthread/src/klibc/kstring.c", "riscv/rtthread/components/drivers/core/device.c" ] LD_FILE="riscv/riscv.ld" TARGET="riscv" OUTPUT="output" def tran_path(path:str): path=os.path.normpath(os.path.join(OUTPUT,path)) base_path=os.path.dirname(path) if not os.path.exists(base_path): os.makedirs(base_path) return path # 判断是否需要重新生成 def check_rebuild(dst:str,src:list): if(not os.path.exists(dst)): return True dst_time=os.path.getmtime(dst) src_time=[] for i in src: src_time.append(os.path.getmtime(i)) src_time.sort() if(src_time[-1]>dst_time): return True return False # 读取.d文件，返回依赖文件列表 def read_depend_files(name:str): with open(name) as f: lines=f.readlines() t='' for line in lines: line=line.strip() if(line[-1]=='\\'): t+=line[:-1] else: t+=line t=t.split(':')[-1].strip() t=t.split(' ') return t @dataclasses.dataclass class cmd_item_t: cmd:str info:str def run_cmd(cmd_queue:Queue,cpu_index:int,return_list:Queue,failed_num): while(not cmd_queue.empty()): if(failed_num.value>0): return try: cmd=cmd_queue.get_nowait() except Exception: return_list.put((cpu_index,True)) return print(f"[{cpu_index}] {cmd.info}") ret=os.system(cmd.cmd) if(ret): return_list.put((cpu_index,False)) failed_num.value+=1 return return_list.put((cpu_index,True)) def run_cmd_queue(cmd_queue:Queue,cpu_num:int=cpu_count()): if(cmd_queue.empty()): return process_list = [] return_list=Queue() failed_num=Value('i',0) for i in range(cpu_num): p = Process(target=run_cmd,args=(cmd_queue,i,return_list,failed_num,)) p.start() process_list.append(p) for i in process_list: i.join() # 消耗掉所有数据防止进程无法退出 while not cmd_queue.empty(): cmd_queue.get() while not return_list.empty(): i=return_list.get() if(not i[1]): print(f"子进程 [{i[0]}] 运行失败") sys.exit(-1) # 保证目标都存在 def check_exists(src:list): for item in src: for i in range(10): if(os.path.exists(item)): break time.sleep(0.1) # 生成依赖关系 def build_depend(src:list): CmdQueue=Queue() dst_list=[] flags=f"{' '.join(INC)} {' '.join(DEF)} {' '.join(CFLAG)}" for i in src: name=os.path.splitext(tran_path(i))[0] dst='.'.join([name,'d']) if(check_rebuild(dst,[i])): cmd=f"{CC} -MM {i} -o {dst} {flags}" CmdQueue.put(cmd_item_t(cmd,f"更新 {dst}")) dst_list.append(dst) run_cmd_queue(CmdQueue) # 生成中间文件 def build_object(src:list): CmdQueue=Queue() dst_list=[] flags=f"{' '.join(INC)} {' '.join(DEF)} {' '.join(CFLAG)}" for i in src: name_t=os.path.splitext(tran_path(i)) name=name_t[0] file_type=name_t[-1] dst='.'.join([name,'o']) cd='.'.join([name,'d']) cmd = '' if(file_type in ['.c','.C']): if(check_rebuild(dst,read_depend_files(cd))): cmd=f"{CC} -c {i} -o {dst} {flags}" elif(file_type in ['.s','.S','.asm','.ASM']): if(check_rebuild(dst,read_depend_files(cd))): cmd=f"{CC} -c {i} -o {dst} {flags}" if(len(cmd)>0): CmdQueue.put(cmd_item_t(cmd,f"编译 {dst}")) dst_list.append(dst) run_cmd_queue(CmdQueue) # 生成可执行文件 def build_target(src:list): flags=f"{' '.join(INC)} {' '.join(DEF)} {' '.join(CFLAG)}" obj_list=[] for i in src: name=os.path.splitext(tran_path(i))[0] obj_list.append('.'.join([name,'o'])) dst=os.path.join(OUTPUT,TARGET)+".elf" if(check_rebuild(dst,obj_list)): cmd=f"{CC} {' '.join(obj_list)} -o {dst} {flags} -T{LD_FILE} \ -Wall -Wextra -nostartfiles -Wl,-Map,\"{OUTPUT}/{TARGET}.map\" \ -Wl,-print-memory-usage" print(f"链接 {dst}") ret=os.system(cmd) if(ret): sys.exit() def main(): global SRC if not os.path.exists(OUTPUT): os.mkdir(OUTPUT) build_depend(SRC) build_object(SRC) build_target(SRC) os.system(f"{OBJCPY} -O binary {OUTPUT}/{TARGET}.elf {OUTPUT}/{TARGET}.bin") os.system(f"{OBJCPY} -O ihex {OUTPUT}/{TARGET}.elf {OUTPUT}/{TARGET}.hex") os.system(f"{OBJDUMP} -d {OUTPUT}/{TARGET}.elf > {OUTPUT}/{TARGET}.lst") if __name__ == "__main__": tick_start=time.time() main() tick_end=time.time() print(f"cost: {tick_end-tick_start}")