Thread Safety

[clintonTE]

Not sure if this is a feature request or a bug report. It is unclear to me what operations pertaining to DataFrames, if any, are thread-safe. For example, I would have expected the below code to be thread-safe since each thread is operating on different parts of the memory, yet it explodes in a cloud of corruption:

using DataFrames
#WARNING: DO NOT RUN THIS

function tsmwecorrupt(N=100_000)
  df = DataFrame(rand(N,100))
  df.grpcol = (i->i%50).(1:N)

  Threads.@threads for sdf ∈ groupby(df, :grpcol)
    sdf.x3 .= -1.
  end

  println(sum(df.x3))
end

tsmwecorrupt()

OUTPUT: 
(many pages of garbage)

On the other hand, this code seems fine:

function tsmwe(N=100_000)
  df = DataFrame(rand(N,100))
  df.grpcol = (i->i%50).(1:N)

  Threads.@threads for r ∈ eachrow(df)
    r.x3 = -1.
  end

  println(sum(df.x3))
end

tsmwe()

OUTPUT: 
-100000.0

and so does this:

function tsmwe2(N=100_000)
  df = DataFrame(rand(N,100))
  df.grpcol = (i->i%50).(1:N)

  Threads.@threads for sdf ∈ collect(groupby(df, :grpcol))
    sdf.x3 .= -1.
  end

  println(sum(df.x3))
end


tsmwe2()

OUTPUT: 
-100000.0

If this is not a bug, then perhaps this issue can serve as a feature request for thread safety under a wider variety of use cases.

EDIT: Also see 1896

[bkamins]

This is a problem with Threads.@threads which assumes that it iterates over a range or an array and nothing else is allowed.

Check out this code as an example:

julia> Threads.@threads for c in "ssf"
       end

Error thrown in threaded loop on thread 0: MethodError(f=typeof(Base.unsafe_getindex)(), args=("ssf", 1), world=0x000000000000643a)

it fails for the same reason.

The way you should fix the code is to write:

  g = groupby(df, :grpcol)
  Threads.@threads for i in 1:length(g)
    sdf = g[i]
    sdf.x3 .= -1.
  end

as now you iterate over a range.

I am leaving this open as @nalimilan who is working on groupby infrastructure might consider:

1. adding axes to GroupedDataFrame object (now we do not define it)
2. and even making a GroupedDataFrame an abstract array (then your code would work); but this is unlikely - I assume that his intention is to rather make it an AbstractDataFrame in the future (but we can discuss it)

[clintonTE]

Got it, makes sense- thanks.

[nalimilan]

This is a problem with Threads.@threads which assumes that it iterates over a range or an array and nothing else is allowed.

I'm not familiar with @threads, so can you explain why it crashes in the provided example? If we defined axes, would it use it? We should probably define it anyway (but not make GroupedDataFrame an AbstractVector since we want e.g. map to return another GroupedDataFrame, not a vector).

EDIT: cross-ref https://discourse.julialang.org/t/threaded-for-loop-over-arbitrary-indexable-objects/26957/1

[bkamins]

If we defined axes, would it use it? We should probably define it anyway

It would not use it, but I think we should define axes as then it would be possible to broadcast over GroupedDataFrame.

so can you explain why it crashes in the provided example?

Threads.@threads expands to use Base.unsafe_getindex internally, and you have:

julia> methods(Base.unsafe_getindex)
# 4 methods for generic function "unsafe_getindex":
[1] unsafe_getindex(r::StepRangeLen{T,#s72,#s71} where #s71<:Base.TwicePrecision where #s72<:Base.TwicePrecision, i::Integer) where T in Base at twiceprecision.jl:447

[2] unsafe_getindex(r::StepRangeLen{T,R,S} where S where R, i::Integer) where T in Base at range.jl:644
[3] unsafe_getindex(r::LinRange, i::Integer) in Base at range.jl:654
[4] unsafe_getindex(A::AbstractArray, I...) in Base at abstractarray.jl:930

and its definition is simply:

function unsafe_getindex(A::AbstractArray, I...)
    @_inline_meta
    @inbounds r = getindex(A, I...)
    r
end

(for the most common case)
so essentially we could define unsafe_getindex ourselves to make it work, but first I want to clarify what core devs want in Base (that is why I have asked on Discourse)

we want e.g. map to return another GroupedDataFrame, not a vector

Actually it would be easy to override map and moreover it would be very simple to override broadcasting to return a GroupedDataFrame (so that transformation.(grouped_data_frame) returns a GroupedDataFrame). This is a more complex decision though so I would defer it (but we can safely add axes now if we want).

[nalimilan]

OK. Actually I've realized that if we define size we will get axes for free. But I'm not sure whether we should do that, as it means GroupedDataFrame would not be just a collection, but a kind of array, which isn't really the case. We could support axes anyway since tuples also support it (even if they don't support size), and it's associated with getindex. Anyway, it looks like @threads should really allow for any indexable, not just arrays.

Regarding broadcast, I guess we could allow it, but it would basically just be a shorthand for map. So maybe better wait until we're sure we don't want to do something else with it (no idea what).

[bkamins]

And I just realized that somehow broadcasting does not requite axes 😄:

julia> describe.(gdf)
3-element Array{DataFrame,1}:
 4×8 DataFrame. Omitted printing of 2 columns
│ Row │ variable │ mean     │ min      │ median   │ max      │ nunique │
│     │ Symbol   │ Float64  │ Float64  │ Float64  │ Float64  │ Nothing │
├─────┼──────────┼──────────┼──────────┼──────────┼──────────┼─────────┤
│ 1   │ x1       │ 0.669556 │ 0.669556 │ 0.669556 │ 0.669556 │         │
│ 2   │ x2       │ 0.375771 │ 0.375771 │ 0.375771 │ 0.375771 │         │
│ 3   │ x3       │ 0.978432 │ 0.978432 │ 0.978432 │ 0.978432 │         │
│ 4   │ x4       │ 0.786694 │ 0.786694 │ 0.786694 │ 0.786694 │         │
 4×8 DataFrame. Omitted printing of 2 columns
│ Row │ variable │ mean      │ min       │ median    │ max       │ nunique │
│     │ Symbol   │ Float64   │ Float64   │ Float64   │ Float64   │ Nothing │
├─────┼──────────┼───────────┼───────────┼───────────┼───────────┼─────────┤
│ 1   │ x1       │ 0.450037  │ 0.450037  │ 0.450037  │ 0.450037  │         │
│ 2   │ x2       │ 0.0699309 │ 0.0699309 │ 0.0699309 │ 0.0699309 │         │
│ 3   │ x3       │ 0.900284  │ 0.900284  │ 0.900284  │ 0.900284  │         │
│ 4   │ x4       │ 0.72257   │ 0.72257   │ 0.72257   │ 0.72257   │         │
 4×8 DataFrame. Omitted printing of 2 columns
│ Row │ variable │ mean     │ min      │ median   │ max      │ nunique │
│     │ Symbol   │ Float64  │ Float64  │ Float64  │ Float64  │ Nothing │
├─────┼──────────┼──────────┼──────────┼──────────┼──────────┼─────────┤
│ 1   │ x1       │ 0.934747 │ 0.934747 │ 0.934747 │ 0.934747 │         │
│ 2   │ x2       │ 0.43983  │ 0.43983  │ 0.43983  │ 0.43983  │         │
│ 3   │ x3       │ 0.753521 │ 0.753521 │ 0.753521 │ 0.753521 │         │
│ 4   │ x4       │ 0.34014  │ 0.34014  │ 0.34014  │ 0.34014  │         │

(although the docs says it should)

[nalimilan]

Funny. It works because it falls back to collect(gdf). I don't really like that, since that means any future change will be breaking, even if we didn't intent this to work. Maybe we should just throw an error for now?

[bkamins]

See #1907

[bkamins]

Reopening.

[bkamins]

You mean that we should disallow collect for GroupedDataFrame? What would be a benefit of this? We support iterate on it.

[nalimilan]

I think I meant disallowing broadcasting on GroupedDataFrame, which #1907 did.

Though we still need to get @threads in Julia to work on custom types?