问题[multithreading](coding)

因此，我偶然发现了 Jeff Preshing 的精彩博客文章，其中介绍了什么是AcquireCPU障碍Release以及如何通过一些 CPU障碍实现这些障碍。

我还读过关于某种总顺序的文章，它保证与非一致性后关系SeqCst一致- 尽管有时由于历史原因它可能与由简单/操作建立的前发生关系相矛盾。AcquireRelease

我的问题是，旧的 GCC 内置函数如何映射到 C++11 (及更高版本) 提出的内存模型？具体来说，如何映射__sync_synchronize()到 C++11 或更高版本的现代 C/C++？

在 GCC手册中，此调用被简单地描述为完整内存屏障，我认为这是所有四种主要屏障（即LoadLoad/ LoadStore/ StoreLoad/StoreStore 屏障）的组合。但sync_synchronize 相当于std::atomic_thread_fence(memory_order_seq_cst)？或者，从形式上讲，其中一个比另一个更强（我想这里就是这种情况：通常，SeqCst隔离应该更强，因为它需要工具链/平台以某种方式即兴执行全局排序，不是吗？），而碰巧大多数 CPU只提供同时满足两者（完整内存屏障__sync_synchronize，完全顺序排序std::atomic_thread_fence(memory_order_seq_cst)）的指令，例如 x86和 PowerPC ？mfencehwsync

__sync_synchronize和要么std::atomic_thread_fence(memory_order_seq_cst)在形式上相等，要么在实际上相等（即，从形式上讲它们是不同的，但没有商业化的 CPU 会费心区分两者），从技术上讲，memory_order_relaxed对同一原子的负载仍然不能依赖于与它同步/创建先发生关系，不是吗？

也就是说，从技术上讲，所有这些断言都可以失败，对吗？

// Experiment 1, using C11 `atomic_thread_fence`: assertion is allowed to fail, right?

// global
static atomic_bool lock = false;
static atomic_bool critical_section = false;

// thread 1
atomic_store_explicit(&critical_section, true, memory_order_relaxed);
atomic_thread_fence(memory_order_seq_cst);
atomic_store_explicit(&lock, true, memory_order_relaxed);

// thread 2
if (atomic_load_explicit(&lock, memory_order_relaxed)) {
    // We should really `memory_order_acquire` the `lock`
    // or `atomic_thread_fence(memory_order_acquire)` here,
    // or this assertion may fail, no?
    assert(atomic_load_explicit(&critical_section, memory_order_relaxed));
}

// Experiment 2, using `SeqCst` directly on the atomic store

// global
static atomic_bool lock = false;
static atomic_bool critical_section = false;

// thread 1
atomic_store_explicit(&critical_section, true, memory_order_relaxed);
atomic_store_explicit(&lock, true, memory_order_seq_cst);

// thread 2
if (atomic_load_explicit(&lock, memory_order_relaxed)) {
    // Again we should really `memory_order_acquire` the `lock`
    // or `atomic_thread_fence(memory_order_acquire)` here,
    // or this assertion may fail, no?
    assert(atomic_load_explicit(&critical_section, memory_order_relaxed));
}

// Experiment 3, using GCC built-in: assertion is allowed to fail, right?

// global
static atomic_bool lock = false;
static atomic_bool critical_section = false;

// thread 1
atomic_store_explicit(&critical_section, true, memory_order_relaxed);
__sync_synchronize();
atomic_store_explicit(&lock, true, memory_order_relaxed);

// thread 2
if (atomic_load_explicit(&lock, memory_order_relaxed)) {
    // we should somehow put a `LoadLoad` memory barrier here,
    // or the assert might fail, no?
    assert(atomic_load_explicit(&critical_section, memory_order_relaxed));
}

我已经在 RPi 5 上尝试了这些代码片段，但我没有看到断言失败。是的，这并没有正式证明任何事情，但它也没有阐明__sync_synchronize和之间的区别std::atomic_thread_fence(memory_order_seq_cst)。

在超线程（或 SMT）中，当 CPU 核心的两个线程被换入和换出时，是否会发生上下文切换。

这会被称为上下文切换吗？如果不是，那么它的术语是什么。

请考虑以下用 Kotlin 编写的 DCL 示例：

@Volatile
private var property: String? = null

private val lock = ReentrantLock()

fun singletonValue(): String {
    if (property == null) {
        lock.withLock {
            if (property == null) {
                property = "Hello World"
            }
        }
    }
    return property!!
}

它与二十年前讨论的 Java 对应部分非常相似，当时 Java 内存模型随着从 Java 1.4 到 Java 5 的过渡而得到修正。话虽如此，我完全确信上述代码片段在 JVM 上运行时会正常运行。

现在，让我们用lateinit var：

@Volatile
private lateinit var property: String

private val lock = ReentrantLock()

fun singletonValue(): String {
    if (!::property.isInitialized) {
        lock.withLock {
            if (!::property.isInitialized) {
                property = "Hello World"
            }
        }
    }
    return property
}

它在语义上与原始示例等同吗？它是否保持相同的原子性和可见性保证？

背景

我有一个Actor名为akka 的程序Client，它管理http与https服务器的连接。客户端有许多功能，包括ping服务和tokenFetcher服务。

代表client一个服务器与另一个服务器之间的“连接”。其设计目的是允许一个服务器server与另一个服务器聊天。

其过程client如下：

1. 定期检查ping对方server是否在线
2. 如果对方server在线，则执行auth并获取令牌
3. 如果令牌有效，则清除向我们请求的所有呼叫

我正在努力解决的是第 3 步。我想知道如何在线程 (actor) 之间安全地实现这一点。

我尝试过的：

我正在使用Seq客户端存储的消息，如下所示：

case class SendApiCall(apiCall: ApiCall, route: String, var sent: Boolean = false)

class Client(server: Server) extends Actor {
    private var apiCalls: Seq[SendApiCall] = Seq.empty[SendApiCall]

    ...

    override def receive: Receive = {
        case sendApiCall@SendApiCall(_, _, _) =>
            if (server.onlineStatus == OFFLINE) {
                apiCalls = apiCalls.appended(sendApiCall)
            }
            else {
                sendApiCall(sendApiCall)
            }
        
        case ServerOnline() => // <- this is send to us from the ping service when it first detects the server is online
            
            apiCalls.iterator.foreach( apiCallRequest =>
                if (!apiCallRequest.sent) {
                    sendApiCall(apiCallRequest)
                    apiCallRequest.sent = true
                }
                apiCallRequest
            )

            apiCalls = apiCalls.filterNot(apiCallRequest => apiCallRequest.sent)
    }
}

但是，我相信apiCalls这是mutable这种情况的状态吗？我想知道：

1. 这个线程安全吗？
2. 如果不是线程安全的，我该如何让它变得安全？

我有一个自制的迭代器，我想循环它，在线程池中执行一些昂贵的处理，并按输入顺序收集结果。

我使用这个链：
Iterator > enumerate() > rayon par_bridge() > map()

当我使用此链并将collect()其放入向量中时，没有任何问题，但当我使用 for 循环时，我收到错误，提示某些内容不是迭代器。
我不明白其中的区别。

这是 MRE：

use std::vec::IntoIter;
use rayon::prelude::*;

fn main() {
    // collecting();
    for_loop();
}

fn collecting() {
    let some_vec = (1..100).collect::<Vec<u8>>();
    let x_iter = AA {y: some_vec.into_iter()};
    
    let with_index: Vec<(usize, u8)> = x_iter
        .enumerate()
        .par_bridge()
        .map(|(i, x)| {(i, x - 1)})
        .collect();

    println!("{:?}", with_index);
}

fn for_loop() {
    let some_vec = (1..100).collect::<Vec<u8>>();
    let x_iter = AA {y: some_vec.into_iter()};

    let mut with_index: Vec<(usize, u8)> = Vec::with_capacity(99);
    for x in x_iter
        .enumerate()
        .par_bridge() // with this line commented out it works fine
        .map(|(i, x)| {(i, x - 1)})
    {
        with_index.push(x);
    }
    println!("{:?}", with_index);
}


struct AA {
    y: IntoIter<u8>
}

impl Iterator for AA {
    type Item = u8;

    fn next(&mut self) -> Option<u8> {
        self.y.next()
    }
}

该collecting函数运行正常。
该for_loop函数给出以下错误：

rayon::iter::Map<IterBridgestd::iter::Enumerate<AA>, {closure@src/main.rs:30:14: 30:22}> 不是迭代器

请解释我为什么会收到这个错误。

我正在尝试在 Delphi 10.4 (Sydney) 中实现多线程。我使用 TTask 并发运行任务，使用 TThreadPool 限制同时运行的线程。

这是我的代码：

var
  MyThreadPool: TThreadPool;
  I: Integer;
  Tasks: TArray<ITask>;

function ProcessTask(const TaskId: Integer): TProc;
begin
  Result := procedure
  Begin
    WriteLn(Format('Started TaskId: %d ThreadId: %d',[TaskId,TThread.Current.ThreadID]));
    Sleep(2000); // Simulate work
    WriteLn(format('TaskId %d completed. ThreadId: %d',[TaskId, TThread.Current.ThreadID]));
  End;
end;

begin
  try

    Writeln('Creating TThreadPool');
    MyThreadPool := TThreadPool.Create;

    try
      MyThreadPool.SetMinWorkerThreads(1);
      MyThreadPool.SetMaxWorkerThreads(4);

      // Hold 10 tasks
      SetLength(Tasks,10);

      for I := 0 to High(Tasks) do
      Begin
        Tasks[I] := TTask.Create(
          ProcessTask(I+1),
          MyThreadPool
        );

        // Start the task
        Tasks[I].Start;
      End;

      // Wait for all tasks to complete
      for I := 0 to High(Tasks) do
        Tasks[I].Wait;

      WriteLn('All tasks completed.');
      ReadLn;

    finally
      MyThreadPool.Free;
    end;

  except
    on E: Exception do
      Writeln(E.ClassName, ': ', E.Message);
  end;
end.

上述代码的输出是：

Creating TThreadPool
Started TaskId: 1 ThreadId: 17292
Started TaskId: 1 ThreadId: 17292
Started TaskId: 2 ThreadId: 19256
Started TaskId: 3 ThreadId: 25268
Started TaskId: 3 ThreadId: 25268
Started TaskId: 4 ThreadId: 21112
TaskId 1 completed. ThreadId: 17292
Started TaskId: 5 ThreadId: 17292
TaskId 2 completed. ThreadId: 19256
Started TaskId: 6 ThreadId: 19256
TaskId 3 completed. ThreadId: 25268
TaskId 3 completed. ThreadId: 25268
TaskId 4 completed. ThreadId: 21112
Started TaskId: 7 ThreadId: 21112Started TaskId: 8 ThreadId: 25268Started TaskId: 7 ThreadId: 21112Started TaskId: 8 ThreadId: 25268
TaskId 5 completed. ThreadId: 17292
Started TaskId: 9 ThreadId: 17292
TaskId 6 completed. ThreadId: 19256
Started TaskId: 10 ThreadId: 19256
TaskId 8 completed. ThreadId: 25268
TaskId 8 completed. ThreadId: 25268
TaskId 7 completed. ThreadId: 21112
TaskId 9 completed. ThreadId: 17292
TaskId 10 completed. ThreadId: 19256
All tasks completed.

代码的问题在于在某些情况下开始和完成报告会重复：

1. Started TaskId出现了两次，分别为 TaskId 1,3,7,8。
2. TaskId完成出现了两次，分别是 TaskId 3,8。

我已经使用 Delphi 有一段时间了，现在正尝试在我的代码中实现多线程，要求如下：

1. 每次只能运行 2 个线程。

2. 其他任务应在队列中等待，并在线程可用时执行。

我尝试使用TParallel.For来实现这一点，但无法实现所需的行为。这是我的示例代码：

    type
       TMyRecord = record
         msg: String;
         sleep: Integer;
          end;

    var
      Form2: TForm2;

    implementation

    {$R *.dfm}

    procedure TForm2.Button1Click(Sender: TObject);
    var
      record1, record2, record3, record4, record5: TMyRecord;
      MyList: TList<TMyRecord>;
    begin
      // Initialize records
      record1.msg := 'Item1';
      record1.sleep := 10;

      record2.msg := 'Item2';
      record2.sleep := 10;

      record3.msg := 'Item3';
      record3.sleep := 3000;

      record4.msg := 'Item4';
      record4.sleep := 3000;

      record5.msg := 'Item5';
      record5.sleep := 100;

     MyList := TList<TMyRecord>.Create;
      try
        // Add records to the list
        MyList.Add(record1);
        MyList.Add(record2);
        MyList.Add(record3);
        MyList.Add(record4);
        MyList.Add(record5);

    // Use TParallel.For
    TParallel.For(0, MyList.Count - 1,
      procedure (i: Integer)
      var
        ListItem: TMyRecord;
      begin
        ListItem := MyList[i];

        TThread.Synchronize(nil,
          procedure
          begin
            Memo1.Lines.Add(ListItem.msg);
          end);

        Sleep(ListItem.sleep);
      end);

      finally
    MyList.Free;
  end;
end;

我也尝试使用TTask.Run和ThreadPool，但无法实现所需的行为。更新 - 这是我尝试使用 Threadpool 执行的操作

procedure TForm2.Button1Click(Sender: TObject);
var
  record1, record2, record3, record4, record5: TMyRecord;
  MyList: TList<TMyRecord>;
  ThreadPool: TThreadPool;
begin
  // Initialize records
  record1.msg := 'Item1';
  record1.sleep := 10;

  record2.msg := 'Item2';
  record2.sleep := 10;

  record3.msg := 'Item3';
  record3.sleep := 3000;

  record4.msg := 'Item4';
  record4.sleep := 3000;

  record5.msg := 'Item5';
  record5.sleep := 100;

  MyList := TList<TMyRecord>.Create;
  try
    // Add records to the list
    MyList.Add(record1);
    MyList.Add(record2);
    MyList.Add(record3);
    MyList.Add(record4);
    MyList.Add(record5);

    ThreadPool := TThreadPool.Create;
    ThreadPool.SetMaxWorkerThreads(2);

    // Iterate through the list and queue each work item with a copy of the record
    for var i := 0 to MyList.Count - 1 do
    begin
      var ListItem := MyList[i];

      ThreadPool.QueueWorkItem(
        procedure
        var
          LocalItem: TMyRecord;
        begin
          LocalItem := ListItem;

          TThread.Queue(nil,
            procedure
            begin
              Memo1.Lines.Add(LocalItem.msg);
              Sleep(LocalItem.sleep);  // Simulate the delay
            end);
        end);
    end;
  finally
    MyList.Free;
  end;
end;

此代码有问题-

1. 它仅打印“item5”5次。
2. 等待时间（睡眠）不起作用。

我该如何修改此代码以将同时运行的线程数限制为 2？并在线程可用时将其他任务排队执行？

我正在尝试使用布尔标志来表示线程应该退出。但是，目前我的代码违反了借用检查器规则。我理解它为什么违反这些规则，但我不知道最合适的设计是什么以及如何解决问题。

fn long_running(exit_flag: &bool) {

    loop {
        std::thread::sleep(std::time::Duration::from_secs(10));

        if *exit_flag {
            break;
        }
    }
}

fn main() {

    let mut exit_flag = false;
    std::thread::scope(
        |scope| {
            let handle = scope.spawn(
                || {
                    long_running(&exit_flag); # (1)
                }
            );

            exit_flag = true; # (2)

            // terminate the consumer poll loop
            handle.join().expect("failed to join thread");
        }
    );
}

问题涉及标有# (1)和 的行# (2)。

• # (1)：引用exit_flag自
• # (2)：在引用仍被借用时尝试改变底层变量

由于借用检查规则，这是不允许的。

• 解决该问题的一个方法可能是堆分配标志，并使用它在两个线程之间共享数据。
• 但是，使用的目的std::thread::scope是允许启动一个可以从父（主）线程的堆栈访问局部变量的线程。

顺便说一下，这个想法是从这个帖子里得到的

• 如何将堆栈变量的引用传递给线程？

如果我在 Raylib 中不使用线程加载纹理，它可以正常工作：

INFO: FILEIO: [assets/spades/5_spades.png] File loaded successfully
INFO: IMAGE: Data loaded successfully (112x176 | R8G8B8 | 1 mipmaps)
INFO: TEXTURE: [ID 46] Texture loaded successfully (112x176 | R8G8B8 | 1 mipmaps)

如果我调用该函数在线程内加载纹理，则会失败：

INFO: FILEIO: [assets/spades/7_spades.png] File loaded successfully
INFO: IMAGE: Data loaded successfully (112x176 | R8G8B8 | 1 mipmaps)
WARNING: TEXTURE: Failed to load texture

但它没有说明为什么加载纹理失败。

我已在此处发布了源代码： https： //github.com/imekon/RaylibThreading

我在平台上看到了很多关于使用线程更新 ProgressBar 的代码，但所有这些都与执行某些计算或更新某些控件属性有关。我不知道这是否可行，但我需要的是 ProgressBar 显示创建 225 (15 X 15) 到 2500 (50 x 50) 个 TextField 的进度。以下是文件和 .fxml。当我包含线程代码部分时，网格停止出现。提前谢谢您。

应用程序

网格应用程序.java

import javafx.application.Application;
import javafx.fxml.FXMLLoader;
import javafx.scene.Scene;
import javafx.stage.Stage;

import java.io.IOException;

public class GridCenterApplication extends Application {
    @Override
    public void start(Stage stage) throws IOException {
        FXMLLoader fxmlLoader = new FXMLLoader(GridCenterApplication.class.getResource("main-view.fxml"));
        Scene scene = new Scene(fxmlLoader.load());
        stage.setTitle("Grid");
        stage.setScene(scene);
        stage.setMaximized(true);
        stage.show();
    }

    public static void main(String[] args) {
        launch();
    }
}

网格控制器.java

import javafx.event.ActionEvent;
import javafx.fxml.FXML;
import javafx.fxml.Initializable;
import javafx.geometry.Pos;
import javafx.scene.control.*;
import javafx.scene.layout.GridPane;

import java.net.URL;
import java.util.ResourceBundle;

public class GridCenterController implements Initializable {

    @FXML
    private ScrollPane scpGrid;

    @FXML
    private Spinner<Integer> spnCols;

    @FXML
    private Spinner<Integer> spnRows;

    @FXML
    private ProgressBar pgbProgress;

    GridPane gridPane;

    private final int CELL_HORIZONTAL_GAP = 1;
    private final int CELL_VERTICAL_GAP = 1;
    private final int CELL_HORIZONTAL_SIZE = 40;
    private final int CELL_VERTICAL_SIZE = 40;

    private int totalCols = 0;
    private int totalRows = 0;

    @FXML
    void onMnuItemNewGridAction(ActionEvent event) {

        if(!(scpGrid.getContent() == null)){
            scpGrid.setContent(null);
        }

        totalCols = spnCols.getValue();
        totalRows = spnRows.getValue();

        var newGrid = new Grid(totalCols, totalRows, CELL_HORIZONTAL_GAP, CELL_VERTICAL_GAP, CELL_HORIZONTAL_SIZE,
                CELL_VERTICAL_SIZE, pgbProgress);
        gridPane = newGrid.getGrid();
        scpGrid.setContent(gridPane);
    }

    @Override
    public void initialize(URL url, ResourceBundle resourceBundle) {
        SpinnerValueFactory<Integer> numberOfCols = new SpinnerValueFactory.IntegerSpinnerValueFactory(15, 50);
        SpinnerValueFactory<Integer> numberOfRows = new SpinnerValueFactory.IntegerSpinnerValueFactory(15, 50);
        spnCols.setValueFactory(numberOfCols);
        spnRows.setValueFactory(numberOfRows);
        scpGrid.contentProperty().addListener((observableValue, oldValue, newValue) -> {
            if (newValue != null && newValue.isVisible()) {
               pgbProgress.setProgress(0);
            }
        });

    }
}

网格.java

import javafx.concurrent.Task;
import javafx.scene.control.ProgressBar;
import javafx.scene.control.TextField;
import javafx.scene.layout.GridPane;

public class Grid {

    private final GridPane grid;

    public Grid(int totalCols, int totalRows, int CELL_HORIZONTAL_GAP, int CELL_VERTICAL_GAP, int CELL_HORIZONTAL_SIZE,
                int CELL_VERTICAL_SIZE, ProgressBar pgbProgress) {

        grid = new GridPane();
        grid.setHgap(CELL_HORIZONTAL_GAP);
        grid.setVgap(CELL_VERTICAL_GAP);
        TextField[][] arrayLetterField = new TextField[totalCols][totalRows];

        Task<Void> task = new Task<Void>() {
            @Override
            protected Void call() throws Exception {

                pgbProgress.setProgress(0);
                double total = totalCols * totalRows;
                double i = 1.0;

                for (int row = 0; row < totalRows; row++) {
                    for (int col = 0; col < totalCols; col++) {
                        arrayLetterField[col][row] = new TextField();
                        arrayLetterField[col][row].setMinSize(CELL_HORIZONTAL_SIZE, CELL_VERTICAL_SIZE);
                        arrayLetterField[col][row].setMaxSize(CELL_HORIZONTAL_SIZE, CELL_VERTICAL_SIZE );
                        grid.add(arrayLetterField[col][row], col, row);
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }

                        updateProgress(i, total);
                        i++;
                    }
                }
                return null;
            }
        };

        pgbProgress.progressProperty().bind(task.progressProperty());
        new Thread(task).start();

    }

    public GridPane getGrid() {
        return grid;
    }

}

主视图.fxml

<?xml version="1.0" encoding="UTF-8"?>

<?import javafx.scene.control.Label?>
<?import javafx.scene.control.Menu?>
<?import javafx.scene.control.MenuBar?>
<?import javafx.scene.control.MenuItem?>
<?import javafx.scene.control.ProgressBar?>
<?import javafx.scene.control.ScrollPane?>
<?import javafx.scene.control.Spinner?>
<?import javafx.scene.layout.BorderPane?>
<?import javafx.scene.layout.Pane?>
<?import javafx.scene.layout.VBox?>

<BorderPane prefHeight="767.0" prefWidth="1053.0" xmlns="http://javafx.com/javafx/17" xmlns:fx="http://javafx.com/fxml/1" fx:controller="com.example.gridpanetest.GridCenterController">
    <top>
        <VBox prefWidth="100.0" BorderPane.alignment="CENTER">
            <children>
                <MenuBar fx:id="mnuBar" prefHeight="25.0" prefWidth="360.0">
                    <menus>
                        <Menu mnemonicParsing="false" text="Grid">
                            <items>
                                <MenuItem mnemonicParsing="false" onAction="#onMnuItemNewGridAction" text="New grid" />
                            </items>
                        </Menu>
                    </menus>
                </MenuBar>
            <Pane prefHeight="80.0" prefWidth="1053.0">
               <children>
                  <Label layoutX="26.0" layoutY="15.0" text="Columns" />
                  <Label layoutX="26.0" layoutY="46.0" text="Rows" />
                  <Spinner fx:id="spnCols" layoutX="79.0" layoutY="11.0" prefHeight="25.0" prefWidth="57.0" />
                  <Spinner fx:id="spnRows" layoutX="79.0" layoutY="42.0" prefHeight="25.0" prefWidth="57.0" />
                  <ProgressBar fx:id="pgbProgress" focusTraversable="false" layoutX="185.0" layoutY="31.0" prefWidth="200.0" progress="0.0" />
               </children>
            </Pane>
            </children>
        </VBox>
    </top>
   <center>
        <ScrollPane fx:id="scpGrid" style="-fx-background-color: #dbbb92; -fx-background: #dbbb92;" BorderPane.alignment="CENTER" />
   </center>
</BorderPane>